MockSchemaRegistryClient not registering avro schema: Cannot get schema from schema registry

user_354161 0
Viewed 2384

I am writing a spring boot Integration test using spring-kafka-test 2.6.3 EmbeddedKafka and Junit 5 for a topology that consumes avro messages. In the test I am using MockSchemaReigstryClient

I am registering the mock schema client and configuring the subjects as suggested in this PR that's now closed. But I am getting and receiving the following error:

    Caused by: org.apache.kafka.common.errors.SerializationException: Error retrieving Avro unknown schema for id 1
Caused by: java.io.IOException: Cannot get schema from schema registry!
    at io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient.getSchemaBySubjectAndIdFromRegistry(MockSchemaRegistryClient.java:176) ~[kafka-schema-registry-client-6.0.1.jar:na]
    at io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient.getSchemaBySubjectAndId(MockSchemaRegistryClient.java:232) ~[kafka-schema-registry-client-6.0.1.jar:na]
    at io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient.getSchemaById(MockSchemaRegistryClient.java:215) ~[kafka-schema-registry-client-6.0.1.jar:na]
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.schemaFromRegistry(AbstractKafkaAvroDeserializer.java:279) ~[kafka-avro-serializer-6.0.1.jar:na]
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:98) ~[kafka-avro-serializer-6.0.1.jar:na]
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:77) ~[kafka-avro-serializer-6.0.1.jar:na]
    at io.confluent.kafka.serializers.KafkaAvroDeserializer.deserialize(KafkaAvroDeserializer.java:55) ~[kafka-avro-serializer-6.0.1.jar:na]
    at io.confluent.kafka.streams.serdes.avro.SpecificAvroDeserializer.deserialize(SpecificAvroDeserializer.java:66) ~[kafka-streams-avro-serde-6.0.1.jar:na]
    at io.confluent.kafka.streams.serdes.avro.SpecificAvroDeserializer.deserialize(SpecificAvroDeserializer.java:38) ~[kafka-streams-avro-serde-6.0.1.jar:na]
    at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:60) ~[kafka-clients-2.6.0.jar:na]
    at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:55) ~[kafka-streams-2.6.0.jar:na]
    at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66) ~[kafka-streams-2.6.0.jar:na]
    at org.apache.kafka.streams.processor.internals.RecordQueue.updateHead(RecordQueue.java:176) ~[kafka-streams-2.6.0.jar:na]
    at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:112) ~[kafka-streams-2.6.0.jar:na]
    at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:185) ~[kafka-streams-2.6.0.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:865) ~[kafka-streams-2.6.0.jar:na]
    at org.apache.kafka.streams.processor.internals.TaskManager.addRecordsToTasks(TaskManager.java:938) ~[kafka-streams-2.6.0.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:640) ~[kafka-streams-2.6.0.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:551) ~[kafka-streams-2.6.0.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:510) ~[kafka-streams-2.6.0.jar:na]

The production code works fine. So, it seems like I am missing something in the test setup. Any pointers will be appreciated. Here's the gist of code.

Update: I am using kafka-schema-registry-client-6.0.1 maven dependency.

confluent-schema-registry confluent-platform apache-kafka-streams apache-kafka
1 Answers

Your StreamsConfig still uses a default SpecificAvroSerde which (yes) uses a MockSchemRegistryClient since you configure the schemaRegistryUrl : mock:// but it does not have the schemas needed for serde. This MockSchemRegistryClient is a different instance than the one you configured for your producer and consumer in your test.

props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde.class.getName());

Consider this flow: (1) Test Produder --> (2) IN_TOPIC -> (3) StreamApp -> (4) OUT_TOPIC -> (5) Test Consumer.

You have configured MockSchemaRegistryClient for Producer at (1), and consumer at (5) and even registered required schemas, but not yet for the one (main app) at (3). Hence the error:

Caused by: java.io.IOException: Cannot get schema from schema registry!

One way to solve this is to use a MockSpecificAvroSerde that points to a registry that has the schemas in test:

application.yaml:

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            configuration:
              specific.avro.reader: true
              default:
                value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde

But for application-test.yaml:

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            configuration:
              schema.registry.url: mock://localtest
              specific.avro.reader: true
              default:
                value.serde: your.custom.serde.that.uses.a.mock.schema.regiter.client.bydefault.MockSpecificAvroSerde
user_49635340
Related