Auto Register Schema

[smeana]

Hello

We are having the following exception in a kafka streams application:

Caused by: io.apicurio.registry.rest.client.exception.ArtifactNotFoundException: No artifact with ID '91' in group 'null' was found.

The serdes are configured to auto register the schema using the same config that we have for our producers or consumers.

        specificSerdeConfig.put(SerdeConfig.REGISTRY_URL, "http://10.8.0.1:8080/apis/registry/v2");
        specificSerdeConfig.put(SerdeConfig.FIND_LATEST_ARTIFACT, Boolean.TRUE);
        specificSerdeConfig.put(SerdeConfig.AUTO_REGISTER_ARTIFACT, Boolean.TRUE);
        specificSerdeConfig.put(SerdeConfig.ENABLE_CONFLUENT_ID_HANDLER, Boolean.TRUE);
        specificSerdeConfig.put(AvroKafkaSerdeConfig.USE_SPECIFIC_AVRO_READER, true);
        specificSerdeConfig.put(SerdeConfig.ARTIFACT_RESOLVER_STRATEGY, RecordIdStrategy.class);

    public static AvroSerde<KeyDay> keyDaySerde() {
        AvroSerde<KeyDay> keyDaySerde = new AvroSerde<>();
        keyDaySerde.configure(specificSerdeConfig, true);
        return keyDaySerde;
    }

    public static AvroSerde<Message> messageSerde() {
        AvroSerde<Message> messageSerde = new AvroSerde<>();
        messageSerde.configure(specificSerdeConfig, false);
        return messageSerde;
    }

The step where it fails is writing-reading to/from a repartition

.groupByKey(Grouped.with(DaySerde(), messageSerde()))
                .reduce(valueServicePrice::aggregateMessage,
                        Materialized.<KeyDay, Message>
                                        as(messagePriceStore)

And even registering it manually is not able to read it.
What group and ID is expecting to do match?

Thanks

[famarting]

Hi, I don't have an specific pointer to what could be going wrong in your application, but I'll point you first to the things I see that could be improved.

First of all, in the current version of our serdes library the usage headers to send the identifier is enabled by default. So the usage of SerdeConfig.ENABLE_CONFLUENT_ID_HANDLER won't make any effect until you set SerdeConfig.ENABLE_HEADERS to false.

Also, you should either use SerdeConfig.FIND_LATEST_ARTIFACT or SerdeConfig.AUTO_REGISTER_ARTIFACT in your configuration. Those properties set the behavior of the serializers, and are mutually exclusive.

Finally, given that you are using RecordIdStrategy the artifact should be created in the groupId equal to the namespace of the avro record and artifactId equal to the avro record name.

I would start by making sure your producers are creating the artifact in the registry whenever you set SerdeConfig.AUTO_REGISTER_ARTIFACT to true , looking at the registry logs should also help troubleshoot the problem.