春云Kafka流:出版到DLQ是失败的与Avro

Question

在使用ErrorHandlingDeserializer处理Avro组合的错误时，我无法发布到Dlq主题。以下是发布时的错误。

主题Topic_DLT在60000毫秒后不在元数据中。错误KafkaConsumerDestination｛consumerDestination Name='Topic‘，partitions=6，dlqName='TOIC_DLT‘｝。container-0-C-1 o.s.i.h.日志处理程序：250-org.springframework.messaging。MessageHandlingException:消息处理程序[org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder中发生错误$1@49abe531]; 嵌套异常为java.lang.RuntimeException:失败，failedMessage=GenericMessage

这是应用程序.yml

spring:
  cloud:
    stream:
      bindings:
        process-in-0:
          destination: TOPIC
          group: groupID
      kafka:
        binder:
          brokers:
            - xxx:9092
          configuration:
            security.protocol: SASL_SSL
            sasl.mechanism: PLAIN
          jaas:
            loginModule: org.apache.kafka.common.security.plain.PlainLoginModule
            options:
              username: username
              password: pwd
          consumer-properties:
            key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
            value.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
            spring.deserializer.value.delegate.class: io.confluent.kafka.serializers.KafkaAvroDeserializer
          producer-properties:
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer            
        bindings:
          process-in-0:
            consumer:
              configuration:
                basic.auth.credentials.source: USER_INFO
                schema.registry.url: registryUrl
                schema.registry.basic.auth.user.info: user:pwd
                security.protocol: SASL_SSL
                sasl.mechanism: PLAIN
              max-attempts: 1
              dlqProducerProperties:
                configuration:
                  basic.auth.credentials.source: USER_INFO
                  schema.registry.url: registryUrl
                  schema.registry.basic.auth.user.info: user:pwd
                key.serializer: org.apache.kafka.common.serialization.StringSerializer
                value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
              deserializationExceptionHandler: sendToDlq
              ackEachRecord: true
              enableDlq: true
              dlqName: TOPIC_DLT
              autoCommitOnError: true
              autoCommitOffhtml" target="_blank">set: true

我使用以下依赖关系:

Spring云依赖 - 2021.0.1 Spring
启动启动器父级 - 2.6.3
Spring云流绑定器Kafka卡模式注册表客户端 - 5.3.0 卡

夫卡序列化程序 - 5.3.0

我不知道我到底错过了什么。

Answer 1

在经历了大量的留档后，我发现为了让Spring完成发布DLQ的工作，我们需要为原始主题和DLT主题拥有相同数量的分区。如果做不到，那么我们需要将dlq分区设置为1或手动提供Dlq分区函数bean。通过提供dlq分区：1，所有消息都将转到分区0。