spring cloud 流 kafka 流 DLQ答案

【问题标题】：spring cloud stream kafka streams DLQspring cloud 流 kafka 流 DLQ
【发布时间】：2021-06-15 19:33:58
【问题描述】：

我正在使用 Apache Kafka 2.7.0 和 Spring Cloud Stream Kafka Streams。

在我的 Spring Cloud Stream (Kafka Streams) 应用程序中，我已将我的 application.yml 配置为在输入主题中的消息出现反序列化错误时使用 sendToDlq 机制：

spring:
  cloud:
    stream:
      function:
        definition: processor      
      bindings:         
        processor-in-0:
          destination: input-topic
          consumer:
            dlqName: input-topic-dlq
        processor-out-0:
          destination: output-topic       
      kafka:
        streams:
            binder:
              deserialization-exception-handler: sendToDlq
            configuration:
              metrics.recording.level: DEBUG
            brokers:
              - localhost:9092

我启动了我的应用程序，但我没有看到此主题存在。文档指出，如果不存在 DLQ 主题，则将创建该主题。

如果我尝试从 DLQ 主题消费，我会收到如下错误：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic input-topic-dlq --property print.value=true --property print.key=true --from-beginning
[2021-03-19 10:17:09,936] WARN [Consumer clientId=consumer-console-consumer-85295-1, groupId=console-consumer-85295] Error while fetching metadata with correlation id 2 : {input-topic-dlq=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

此时，当我查询 Zookeeper ls /brokers/topics 时，我看到创建的主题。

现在，我尝试将非 JSON 消息发布到输入主题（我的默认解串器是 JSON）。

但我在创建的 input-topic-dlq 主题中看不到任何消息。

奇怪的是，我可以在默认的“error.input-topic-dlq.appId”主题中看到消息。

我在这里做错了吗？

【问题讨论】：

我正在努力解决同样的问题。据我所见，spring.cloud.stream.bindings 中提供的消费者属性被解析为没有dlqName 字段的org.springframework.cloud.stream.binder.ConsumerProperties。只有org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties 似乎有它，但它似乎只用于spring.cloud.stream.kafka.bindings 消费者属性。

标签： java spring apache-kafka stream cloud

【解决方案1】：

我设法弄清楚了。 Spring Cloud Stream Kafka Streams Binder 的当前文档中似乎存在拼写错误。

绑定目标应该在spring.cloud.streams.bindings 级别，就像您已经拥有的一样，但特定于实现的消费者属性应该在spring.cloud.streams.kafka.streams.bindings 级别。

所以你的配置应该是这样的：

spring:
  cloud:
    stream:
      function:
        definition: processor      
      bindings:         
        processor-in-0:
          destination: input-topic
        processor-out-0:
          destination: output-topic       
      kafka:
        streams:
            binder:
              deserialization-exception-handler: sendToDlq
            bindings:
              processor-in-0:
                consumer:
                  dlqName: input-topic-dlq
            configuration:
              metrics.recording.level: DEBUG
            brokers:
              - localhost:9092

【讨论】：