如何使用ErrorHandlingDeserializer与Spring云流绑定器kafka将错误记录发送到默认的DLQ主题？

Question

我想使用ErrorHandlingDeserializer来处理反序列化错误，并使用spring cloud stream binder kafka将错误记录发送到kafka binder的默认DLQ主题。

spring:
  application:
    name: my-service
  cloud:
    ## ====================================================== Vault Setting Omitted ====================================================
    vault:
    ## ====================================================== Vault Setting Omitted ====================================================
    schemaRegistryClient:
      enabled: true
      endpoint: my-schema-registry.com
      basic:
        auth:
          # user: This property is resolved from Vault.
          # password: This property is resolved from Vault.
    stream:
      kafka:
        heartbeat:
          interval:
            ms: 1000
        binder:
          brokers: my-kafka-broker.com:9094
          ## =============================================== SerDes ====================================================
          consumer-properties:
            allow.auto.create.topics: false
            specific.avro.reader: true
            schema.registry.url: ${spring.cloud.schemaRegistryClient.endpoint}
            auto.register.schemas: false
            basic.auth.credentials.source: USER_INFO
            basic.auth.user.info: ${spring.cloud.schemaRegistryClient.basic.auth.user}:${spring.cloud.schemaRegistryClient.basic.auth.password}
          producer-properties:
            allow.auto.create.topics: false
            key.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
            value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
            schema.reflection: true
            schema.registry.url: ${spring.cloud.schemaRegistryClient.endpoint}
            auto.register.schemas: false
            basic.auth.credentials.source: USER_INFO
            basic.auth.user.info: ${spring.cloud.schemaRegistryClient.basic.auth.user}:${spring.cloud.schemaRegistryClient.basic.auth.password}
          ## =========================================== Kafka Security Omitted ================================================
          configuration:
          ## =========================================== Kafka Security Omitted ================================================
        bindings:
          my-in-0:
            consumer:
              enableDlq: true
              autoCommitOnError: true
              configuration:
                key.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
                value.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
                spring.deserializer.key.delegate.class: io.confluent.kafka.serializers.KafkaAvroDeserializer
                spring.deserializer.value.delegate.class: io.confluent.kafka.serializers.KafkaAvroDeserializer
              dlqProducerProperties:
                configuration:
                  key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer
                  value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer
      default:
        binder: kafka
        producer:
          useNativeEncoding: true
        consumer:
          useNativeEncoding: true
      function:
        definition: myFunction
      bindings:
        my-in-0:
          binder: kafka
          destination: my-in-topic
          content-type: application/*+avro
          group: myGroup
        my-out-0:
          binder: kafka
          destination: my-out-topic
          content-type: application/*+avro
          group: myGroup

Answer 1

禁用绑定中的dlq设置，并使用配置了死信发布恢复器的SeekTocurErrorHandler配置侦听器容器（使用ListenerContainerCustomizer bean）。

https://docs.spring.io/spring-kafka/docs/current/reference/html/#error-handlers

https://docs.spring.io/spring-kafka/docs/current/reference/html/#dead-信件