SpringBoot Embedded Kafka 使用 Avro Schema

如何解决SpringBoot Embedded Kafka 使用 Avro Schema

我创建了以下测试类来使用 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 的模拟实现。这个版本不是
线程安全。模式数据存储在内存中，不会持久化或跨实例共享。 */