我們將 Kafka 與 Spring Cloud Stream 一起使用，我們需要連接到 Spring Boot 組件中的 Confluent Schema Registry，請參閱https://github.com/donalthurley/KafkaConsumeScsAndConfluent。我們添加了以下配置來創建所需的 ConfluentSchemaRegistryClient bean，請參閱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) AvroSchmaRegistryClientMessageConverter 正在調用 DefaultSchemaRegistryClient 的事實向我們表明 ConfluentSchemaRegistryClient bean 的接線存在問題。我們的配置中是否還需要其他內容來確保 ConfluentSchemaRegistryClient bean 正確連接？
2 回答
ibeautiful
TA貢獻1993條經驗 獲得超6個贊
它對我有用。這就是我所做的：
完全像你一樣使用配置類
@EnableSchemaRegistryClient在項目中使用了注解將 avro 序列化程序添加到類路徑：
io.confluent:kafka-avro-serializer設置屬性如下：
spring.cloud.stream.kafka.bindings.channel.consumer.configuration.schema.registry.url=你的注冊地址 spring.cloud.stream.kafka.bindings.channel.consumer.configuration.specific.avro.reader=true
其中 channel 對應于您應用中的頻道名稱。
我認為最后一個屬性對于告訴 Spring 使用 Avro 序列化器而不是默認序列化器非常重要。
我使用的是 Spring Cloud Stream Elmhurst.RELEASE，因此如果您使用其他版本，屬性的名稱可能會略有不同。
炎炎設計
TA貢獻1808條經驗 獲得超4個贊
我現在已將 @EnableSchemaRegistryClient 注釋移至項目應用程序類，請參閱https://github.com/donalthurley/KafkaConsumeScsAndConfluent/commit/b4cf5427d7ab0a4fed619fe54b042890f5ccb594并重新部署，這解決了我在部署環境時遇到的問題。
我一直在用 @EnableSchemaRegistryClient 注釋來注釋我的生產者和消費者類。
在我所有的本地測試中，這一直在針對我本地的 docker confluent 模式注冊表工作。然而，在部署到我們的環境時，它大部分時間都在工作，但在一些部署后偶爾會失敗。
我沒有成功地在本地復制這個。
我還在本地測試中注意到，如果我刪除 Confluent Schema Registry 的配置，我會得到相同的空指針異常堆棧跟蹤。
所以我認為我看到的問題是，當項目應用程序類中不存在 @EnableSchemaRegistryClient 注釋時，AvroSchemaRegistryClientMessageConverter bean 沒有與融合模式注冊表 bean 連接。
我不明白為什么那是必要的，但我認為它可能已經解決了這個問題。
添加回答
舉報
0/150
提交
取消