未发送到 DLQ 主题的消息答案

【问题标题】：Messages not sent to DLQ topic未发送到 DLQ 主题的消息
【发布时间】：2020-06-17 10:18:15
【问题描述】：

我想使用 DLQ 处理异常

这里是 application.yml 主题已成功创建，但我的 DLQ 主题中没有收到异常消息

spring:
  cloud:
    stream:
      default:
        consumer:
          useNativeEncoding: true
      kafka:
        binder:
          brokers:
            - localhost:9092
          consumer-properties:
            key.deserializer : org.apache.kafka.common.serialization.StringDeserializer
            value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
            schema.registry.url: http://localhost:8081
            specific.avro.reader: true
            enable.auto.commit: true
        bindings:
          resourceInventoryInput:
            consumer:
              autoCommitOffset: true
              autoCommitOnError: true
              enableDlq: true
              dlqName: dead-out
              dlqProducerProperties:
                configuration:
                  key.serializer: org.apache.kafka.common.serialization.StringSerializer
                  value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      bindings:
        resourceInventoryInput:
          binder: kafka
          destination: ${application.messaging.topic}
          content-type: application/*+avro
          group: ${application.messaging.group}
      default-binder: kafka

【问题讨论】：

标签： avro spring-kafka spring-cloud-stream dead-letter

【解决方案1】：

不要在多个地方问同一个问题；这是浪费您和我们的时间。

我已经回复你on GitHub。

我刚刚使用 Boot 2.1.15 和 Greenwich.SR6（以及 Boot 2.2.8/Hoxton.SR5）测试了您的 yaml，它运行良好。我所做的唯一更改是更改绑定名称以输入和注释掉 avro 内容。

@SpringBootApplication
@EnableBinding(Sink.class)
public class Kbgh9181Application {

    public static void main(String[] args) {
        SpringApplication.run(Kbgh9181Application.class, args);
    }

    @StreamListener(Sink.INPUT)
    public void listen(String in) {
        throw new RuntimeException("foo");
    }

    @KafkaListener(id = "kbgh918", topics = "dead-out", properties = "auto.offset.reset:earliest")
    public void listen(Message<?> in) {
        System.out.println(in);
    }

}

GenericMessage [payload=byte[3], headers={x-original-offset=[B@67917b81, x-original-partition=[B@467895cd, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedTopic=dead-out , kafka_offset=5, x-exception-message=[B@51def01e, x-exception-fqcn=[B@531d42e5, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@3fbc6674, x-original-topic=[ B@3d684ab3, x-original-timestamp-type=[B@1b101300, kafka_receivedPartitionId=0, x-original-timestamp=[B@222370ed, kafka_receivedTimestamp=1592402977606, x-exception-stacktrace=[B@7e703d1b}]

【讨论】：