使用默认架构注册表客户端而不是Avro架构注册表客户端解决Spring Cloud Stream问题

Question

我们正在使用Kafka和Spring Cloud Stream，我们需要连接到Spring Boot组件中的Confluent Schema Registry，请参阅https://github.com/donalthurley/KafkaConsumeScsAndConfluent.

我们添加了以下配置来创建所需的 ConfluentSchemaRegistryClient bean see https://github.com/donalthurley/KafkaConsumeScsAndConfluent/blob/master/src/main/java/com/example/kafka/KafkaConfig.java，它应该覆盖 Spring Cloud Stream 中的默认模式注册表。

但是，在某些部署之后，我们间歇性地看到以下故障。

org.springframework.messaging.MessageDeliveryException: failed to send Message to channel

根本原因显示此堆栈跟踪

Caused by: java.lang.NullPointerException
    at     org.springframework.cloud.stream.schema.client.DefaultSchemaRegistryClient.register(DefaultSchemaRegistryClient.java:71)
    at org.springframework.cloud.stream.schema.avro.AvroSchemaRegistryClientMessageConverter.resolveSchemaForWriting(AvroSchemaRegistryClientMessageConverter.java:238)
    at org.springframework.cloud.stream.schema.avro.AbstractAvroMessageConverter.convertToInternal(AbstractAvroMessageConverter.java:179)
    at org.springframework.messaging.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:201)
    at org.springframework.messaging.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:191)
    at org.springframework.messaging.converter.CompositeMessageConverter.toMessage(CompositeMessageConverter.java:83)
    at org.springframework.cloud.stream.binding.MessageConverterConfigurer$OutboundContentTypeConvertingInterceptor.doPreSend(MessageConverterConfigurer.java:322)
    at org.springframework.cloud.stream.binding.MessageConverterConfigurer$AbstractContentTypeInterceptor.preSend(MessageConverterConfigurer.java:351)
    at org.springframework.integration.channel.AbstractMessageChannel$ChannelInterceptorList.preSend(AbstractMessageChannel.java:611)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453)

DefaultSchemaRegistryClient 由 AvroSchemaRegistryClientMessageConverter 调用的事实向我们表明，我们的 ConfluentSchemaRegistryClient bean 的接线存在问题。

在我们的配置中还需要其他东西来确保 ConfluentSchemaRegistryClient Bean 正确连接吗？

Answer 1

现在，我已经将@Enable组分注册客户端注释移动到项目应用程序类（请参见https://github.com/donalthurley/KafkaConsumeScsAndConfluent/commit/b4cf5427d7ab0a4fed619fe54b042890f5ccb594并重新部署），这修复了在我们的环境中部署时遇到的问题。

我一直在用@EnableSchemaRegistryClient注释来注释我的生产者和消费者类。

在我所有的本地测试中，这一直针对我的本地 docker 融合模式注册表工作。但是，在部署到我们的环境时，它大部分时间都在工作，但在一些部署后偶尔会失败。

我没有成功地在本地复制这个。

我还注意到，在本地测试中，如果删除了Confluent Schema Registry的配置，我会得到相同的空指针异常堆栈跟踪。

所以我想我看到的问题是，当项目应用程序类中不存在@EnableSchemaRegistryClient注释时，AvroSchemaRegistryClientMessageConverter bean没有与融合模式注册表bean连接。

我不明白为什么这是必要的，但我认为它可能已经解决了这个问题。

Answer 2

它对我有用。这就是我所做的：

>

使用配置类的方式与您完全相同

在项目中使用了@EnableSchemaRegistryClient注释

将avro序列化程序添加到classpath：＜code＞io。汇流：kafka avro串行器

按如下方式设置属性：

spring.cloud.stream.kafka.bindings.channel.consumer.configuration.schema.registry.url=注册表的地址spring.cloud.stream.kafka.bindings.channel.consumer.configuration.specific.avro.reader=true

其中频道对应于您应用程序中的频道名称。

我认为最后一个属性非常重要，告诉 Spring 使用 Avro 序列化程序而不是默认序列化程序。

我使用的是Spring-Cloud-Stream埃尔姆赫斯特。RELEASE，因此如果使用其他版本，属性的名称可能会略有不同。