MockSchemaRegistryClient 未注册 avro schema:无法从架构注册表获取架构

Question

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

我正在注册模拟架构客户端并按照此 PR 中的建议配置主题，该 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]

生产代码工作正常。所以，似乎我在测试设置中缺少一些东西。任何指示将不胜感激。这是代码的要点。

更新：我正在使用 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) Test Produder --> (2) IN_TOPIC -> (3) StreamApp -> (4) OUT_TOPIC -> (5) Test Consumer。

您已经在 (1) 处为生产者配置了MockSchemaRegistryClient，在 (5) 为使用者配置了，甚至注册了所需的架构，但尚未为 (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