Spring Cloud Stream绑定Pulsar时死信策略未生效问题

依赖与配置背景 #

使用的依赖：

implementation("org.springframework.pulsar:spring-pulsar-spring-cloud-stream-binder:0.1.1-SNAPSHOT")

当前application.yaml配置：

# pulsar
pulsar:
  client:
    service-url: pulsar://localhost:6650
cloud:
  function:
    definition: notificationListener
  stream:
    bindings:
      notificationListener-in-0:
        destination: notification
        consumer:
          use-native-decoding: true

    pulsar:
      bindings:
        notificationListener-in-0:
          consumer:
            negative-ack-redelivery-delay: 1s
            dead-letter-policy:
              dead-letter-topic: notification-dlq
              max-redeliver-count: 5
            schema-type: JSON
            message-type: global.din.notification.data.dto.BrokerMessage

死信策略不生效的排查与解决步骤 #

• 升级依赖版本：0.1.1-SNAPSHOT是早期快照版本，存在配置绑定的已知bug。建议替换为稳定版，例如适配Spring Boot 3.x的spring-pulsar-spring-cloud-stream-binder:2.1.0（版本号可根据Spring Boot版本调整）。

• 确认死信主题已创建：Pulsar默认不会自动生成死信主题，需手动创建notification-dlq，或添加客户端自动创建配置：

  pulsar:
    client:
      service-url: pulsar://localhost:6650
      auto-create-topic: true
      allow-auto-update-partitions: true
  

• 检查YAML缩进格式：确保配置层级正确，所有dead-letter-policy相关配置严格缩进在pulsar.bindings.notificationListener-in-0.consumer下，YAML缩进需统一使用2或4空格，禁止混用制表符和空格。

• 确保异常触发负确认：死信策略仅在消费端抛出未捕获异常、触发负确认时生效。消费函数需避免捕获所有异常，或手动抛出异常触发重试：

  @Bean
  public Consumer<BrokerMessage> notificationListener() {
      return message -> {
          try {
              // 业务处理逻辑
              System.out.println("处理消息：" + message);
          } catch (Exception e) {
              // 抛出异常触发负确认，触发重试与死信
              throw new RuntimeException("消息处理失败", e);
          }
      };
  }
  

• 验证配置加载情况：开启Debug日志查看消费者配置是否正确加载：

  logging:
    level:
      org.springframework.pulsar: DEBUG
      org.apache.pulsar.client.impl.ConsumerImpl: DEBUG
  

  启动后搜索日志中的dead-letter-topic、maxRedeliverCount关键字，确认配置是否被应用到消费者实例。

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