Spring Cloud Stream中DLQ测试失败：payload始终为null的解决方法

问题根源 #

你当前使用的InputDestination和OutputDestination是Spring Cloud Stream的Test Channel Binder组件，它基于内存通道模拟消息流转，并不支持Kafka原生的死信队列（DLQ）特性。你的DLQ配置是针对Kafka绑定器的，Test Channel Binder会忽略这些配置，所以失败消息不会被转发到指定的DLQ队列，导致payload始终为null。

修复方案：使用EmbeddedKafka进行测试 #

要验证Kafka DLQ的功能，必须使用真实的Kafka环境，Spring提供的EmbeddedKafka可以在测试中启动一个内嵌的Kafka实例，配合Kafka绑定器完成测试。

步骤1：添加必要依赖 #

确保你的build.gradle.kts（或Maven）中包含EmbeddedKafka相关依赖：

testImplementation("org.springframework.kafka:spring-kafka-test")
testImplementation("org.springframework.cloud:spring-cloud-stream-test-support")

步骤2：调整测试类配置 #

使用@EmbeddedKafka注解启动内嵌Kafka，同时配置Spring Cloud Stream绑定器指向该内嵌实例：

@SpringBootTest(
    properties = [
        "spring.cloud.stream.kafka.binder.brokers=\${spring.embedded.kafka.brokers}",
        "spring.cloud.stream.kafka.binder.configuration.cache.max.bytes.buffering=0",
        "spring.cloud.stream.bindings.input.group=inflow",
        "spring.cloud.stream.bindings.input.consumer.max-attempts=1",
        "spring.cloud.stream.kafka.bindings.input.consumer.enable-dlq=true",
        "spring.cloud.stream.kafka.bindings.input.consumer.dlq-name=inflow-parsingdlq"
    ]
)
@EmbeddedKafka(partitions = 1, topics = ["input", "inflow-parsingdlq"])
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
class DlqTest {

    @Autowired
    private lateinit var kafkaTemplate: KafkaTemplate<String, Any>

    @Autowired
    private lateinit var consumerFactory: ConsumerFactory<String, Any>

    @Test
    fun testDlqProcessing() {
        // 发送会处理失败的消息到输入主题
        val rawString = readFileAsString("failing_rawstring.json")
        val inflowMessage = InflowMessage(rawString)
        kafkaTemplate.send("input", inflowMessage)

        // 创建DLQ队列的消费者，读取消息
        val consumer = consumerFactory.createConsumer()
        consumer.subscribe(listOf("inflow-parsingdlq"))
        val records = consumer.poll(Duration.ofSeconds(5))

        // 断言DLQ中有消息
        assertThat(records.count()).isGreaterThan(0)
        consumer.close()
    }

    private fun readFileAsString(filename: String): String {
        return this::class.java.classLoader.getResource(filename)?.readText() ?: throw IllegalArgumentException("File not found: $filename")
    }
}

关键配置说明 #

• @EmbeddedKafka：启动内嵌Kafka，提前创建输入主题和DLQ主题，避免主题不存在的问题。
• spring.cloud.stream.kafka.binder.brokers=\${spring.embedded.kafka.brokers}：让Stream绑定器连接到内嵌Kafka实例。
• @DirtiesContext：测试结束后清理上下文，避免后续测试受影响。

额外注意事项 #

• 确保你的parsing函数（定义的Spring Cloud Function）在处理InflowMessage时确实会抛出异常，这样消息才会被转发到DLQ。如果函数没有抛出异常，DLQ不会收到消息。
• max-attempts=1配置确保消息只尝试处理一次就进入DLQ，符合你的初始配置。

常见问题排查 #

如果仍然收不到DLQ消息，检查：

• 函数处理逻辑是否确实抛出了异常。
• 主题名称是否和配置的dlq-name完全一致，Kafka主题名称区分大小写。
• 内嵌Kafka是否正常启动，可以通过日志查看Kafka启动状态。

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