问题描述
我创建了以下测试类来使用 AvroSerializer 生成事件。
@SpringBoottest
@EmbeddedKafka(partitions = 1,brokerProperties = { "listeners=PLAINTEXT://localhost:9092","port=9092" })
@TestPropertySource(locations = ("classpath:application-test.properties"))
@ContextConfiguration(classes = { TestAppConfig.class })
@DirtiesContext
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class EntitlementEventsConsumerServiceImpltest {
@Autowired
EmbeddedKafkabroker embeddedKafkabroker;
@Bean
MockSchemaRegistryClient mockSchemaRegistryClient() {
return new MockSchemaRegistryClient();
}
@Bean
KafkaAvroSerializer kafkaAvroSerializer() {
return new KafkaAvroSerializer(mockSchemaRegistryClient());
}
@Bean
public DefaultKafkaProducerFactory producerFactory() {
Map<String,Object> props = KafkaTestUtils.producerProps(embeddedKafkabroker);
props.put(KafkaAvroSerializerConfig.AUTO_REGISTER_SCHEMAS,false);
return new DefaultKafkaProducerFactory(props,new StringSerializer(),kafkaAvroSerializer());
}
@Bean
public KafkaTemplate<String,ApplicationEvent> kafkaTemplate() {
KafkaTemplate<String,ApplicationEvent> kafkaTemplate = new KafkaTemplate(producerFactory());
return kafkaTemplate;
}
}
但是当我使用 kafkaTemplate().send(appEventsTopic,applicationEvent);
发送事件时,我收到以下异常。
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema Not Found; error code: 404001
at io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient.getIdFromregistry(MockSchemaRegistryClient.java:79)
at io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient.getId(MockSchemaRegistryClient.java:273)
at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:82)
at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:53)
at org.apache.kafka.common.serialization.Serializer.serialize(Serializer.java:62)
at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:902)
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:862)
at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.send(DefaultKafkaProducerFactory.java:781)
at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:562)
at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:363)
当我使用 MockSchemaRegistryClient 时,为什么它会尝试查找架构?
解决方法
您将生产者设置为不尝试在生成消息时自动注册新架构,因此它只是尝试从 SR 中获取数据,而在 SR 上没有找到其架构。
也没有看到您设置架构注册表 URL 猜测其采用默认值
对于您的问题,模拟是模仿真实模式注册表的工作,但有其明显的缺点
/**
- 可用于测试的 SchemaRegistryClient 的模拟实现。这个版本不是
- 线程安全。模式数据存储在内存中,不会持久化或跨实例共享。 */
您可以查看文档以获取更多信息