Spring Cloud Stream批量配置DeadLetterPublishingRecoverer发消息至DLQ

问题根因 #

当前配置出现连接默认localhost:9092的核心原因是：手动注入的KafkaOperations是Spring Kafka默认自动配置生成的客户端，仅读取spring.kafka.*前缀的配置，不会读取Spring Cloud Stream Kafka Binder专属的spring.cloud.stream.kafka.binder.*配置，因此没有加载自定义的broker地址。 #

问题1：配置DLQ读取application.properties中Kafka元数据的方案 #

有两种可落地的实现方式，优先选择第一种内置方案，配置量最小、兼容性最好：

• 方案一：使用Spring Cloud Stream Kafka内置DLQ能力
  直接删除自定义的DefaultErrorHandler、ListenerContainerCustomizer、手动实例化DeadLetterPublishingRecoverer的代码，在配置文件中添加如下配置即可，内置DLQ组件会自动复用binder加载的Kafka连接配置，不会出现地址不匹配问题：

  # 开启消费端DLQ
  spring.cloud.stream.kafka.bindings.consume-in-0.consumer.enable-dlq=true
  # 指定DLQ主题名称
  spring.cloud.stream.kafka.bindings.consume-in-0.consumer.dlq-name=error.topic.name
  # 关闭容器层面的发送重试（适配业务层已实现的Spring Retry逻辑）
  spring.cloud.stream.kafka.bindings.consume-in-0.consumer.dlq-producer-factory.configuration.retries=0
  # 开启批量失败整批投递
  spring.cloud.stream.kafka.bindings.consume-in-0.consumer.batching-dlq=true
  

• 方案二：手动定义错误处理器时复用Binder的Kafka客户端配置
  如果需要保留自定义错误处理逻辑，不要注入默认的KafkaOperations实例，通过BinderFactory获取Kafka Binder内置的、已加载正确配置的Producer工厂来构建DLQ发送用的KafkaTemplate，代码示例：

  @Bean
  public DefaultErrorHandler errorHandler(BinderFactory binderFactory) {
      KafkaMessageChannelBinder kafkaBinder = 
          (KafkaMessageChannelBinder) binderFactory.getBinder("kafka", MessageChannel.class);
      // 用binder的producer工厂构建template，自动继承binder的所有Kafka配置
      KafkaTemplate<Object, Object> dlqTemplate = new KafkaTemplate<>(kafkaBinder.getProducerFactory());
      DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(dlqTemplate,
          (consumerRecord, exception) -> new TopicPartition("error.topic.name", 0));
      // 适配批量消费场景，开启整批投递
      recoverer.setBatchListener(true);
      recoverer.setCommitRecovered(true);
      return new DefaultErrorHandler(recoverer, new FixedBackOff(0, 0));
  }
  

问题2：是否支持通过Supplier函数创建DLQ提供者 #

不支持直接使用普通Supplier函数作为DLQ的发送提供者。
DLQ发送逻辑属于Kafka监听容器错误处理链路的内置环节，触发时机是消费重试耗尽后，由错误处理器直接调用Kafka客户端完成失败消息投递，整个流程不会走Spring Cloud Function的函数绑定、消息路由链路，因此无法通过自定义Supplier函数替换DLQ的发送实现。 #

内容的提问来源于stack exchange，提问作者Gourab27