apache-kafka - MockSchemaRegistryClient 未注册 avro 模式 : Cannot get schema from schema registry

Question

我正在使用 spring-kafka-test 2.6.3 EmbeddedKafka 和 Junit 5 为使用 avro 消息的拓扑编写 spring boot 集成测试。在测试中我使用的是 MockSchemaReigstryClient

我正在注册模拟模式客户端并按照 PR 中的建议配置主题那现在已经关闭了。但是我收到并收到以下错误:

    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]

生产代码运行良好。所以，似乎我在测试设置中遗漏了一些东西。任何指针将不胜感激。这是 gist代码。

更新:我正在使用 kafka-schema-registry-client-6.0.1 maven 依赖项。

Answer 1

您的 StreamsConfig 仍然使用默认的 SpecificAvroSerde，它(是)使用 MockSchemRegistryClient 因为您配置了 schemaRegistryUrl : mock:// 但它没有 serde 所需的模式。此 MockSchemRegistryClient 与您在测试中为生产者和消费者配置的实例不同。

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

考虑以下流程:(1) 测试生产者 --> (2) IN_TOPIC -> (3) StreamApp -> (4) OUT_TOPIC -> (5) 测试消费者。

您已经在 (1) 处为生产者配置了 MockSchemaRegistryClient，在 (5) 处为消费者配置了 MockSchemaRegistryClient，甚至还为 (3) 处的一个(主应用程序)注册了所需的模式。因此错误:

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

解决此问题的一种方法是使用指向具有测试模式的注册表的 MockSpecificAvroSerde:

application.yaml:

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

但是对于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