Spring Cloud Stream Kafka批量模式下DLQ报RecordTooLargeException问题咨询

我们的项目使用Spring Cloud Stream Kafka Binder，开启了批量模式（batch-mode=true），批量大小设置为500条消息，采用DeadLetterPublishingRecoverer将失败消息发送至死信队列（DLQ），此前功能正常。近期出现异常：发送消息至DLQ时，Kafka抛出RecordTooLargeException: Message is 15762601 bytes when serialized which is larger than 1048576。

经排查发现：当单条消息处理失败抛出BatchListenerFailedException时，Kafka会将整批500条消息封装进kafka_dlt-exception-message和kafka_dlt-exception-stacktrace这两个Kafka Header中，导致消息体积远超限制。目前考虑通过移除这两个Header解决问题，想咨询该处理方式是否合理。

使用的Spring Cloud Stream版本为3.2.6。

错误堆栈 #

Caused by: org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1@2a4bc0e4]; nested exception is org.springframework.kafka.listener.BatchListenerFailedException: Exception while generating outbound message; nested exception is java.lang.NullPointerException, failedMessage=GenericMessage [payload=[{"message":{src:"test"}}, {"message":{src:"test1"}}, {"message":{src:"test2"}}, headers={skip-input-typeconversion=false,kafka_offset=[2193, 2194, 2195],kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@4730c07,  kafka_conversionFailures=[], kafka_timestampType=[CREATE_TIME, CREATE_TIME, CREATE_TIME], kafka_receivedPartitionId=[2, 2, 2], kafka_receivedMessageKey=[test, test1, test2], kafka_batchConvertedHeaders=[{}, {}, {}], kafka_receivedTopic=[input-topic, input-topic, input-topic], kafka_receivedTimestamp=[1677649913630, 1677649913629,1677649913629], contentType=application/json, kafka_groupId=group1}]
    at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:191)
    at org.springframework.integration.handler.AbstractMessageHandler.doHandleMessage(AbstractMessageHandler.java:108)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:65)
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)

消费者函数代码 #

@Bean
Consumer<Message<String>> processMessages() {
    return input -> {
        try {
            // Processing Logic
        } catch (CustomException exp) {
            throw new BatchListenerFailedException(exp.getMessage(), exp.getCause(), exp.getIndex()); // getIndex() returns the index of the failed message
        }
    };
}

经调试确认：AbstractMessageHandler的handleMessage方法异常块中，会将整批消息传入MessageHandlingException，而非仅传入失败消息，这导致上述两个Header包含全量批量消息，进而使消息体积剧增。

移除指定Header的合理性 #

这种处理方式完全合理，理由如下：

• 这两个Header的核心作用是记录异常信息，但在批量场景下，整批消息被塞入Header直接导致消息体积超限，触发Kafka的RecordTooLargeException，反而阻碍了死信消息的正常投递，违背了DLQ的设计初衷。
• 你的代码中已经通过BatchListenerFailedException指定了失败消息的索引（exp.getIndex()），结合DLQ中消息本身的内容，已经足够定位到具体的失败消息和问题原因，额外携带整批消息的异常Header属于冗余信息。
• 移除这两个Header不会影响DLQ的核心功能：死信消息依然会被正确投递，你依然可以通过消息本身的内容和业务异常信息排查问题。

其他可选优化方案 #

除了直接移除Header，你还可以考虑以下方案：

• 自定义DeadLetterPublishingRecoverer：重写其构建死信消息的逻辑，只将单条失败消息而非整批消息塞入异常Header，或者对异常信息进行精简（比如只保留异常栈的关键部分）。
• 调整Kafka消息大小限制：临时增大message.max.bytes（Broker端）和max.request.size（Producer端）配置，但这只是治标不治本的方案，随着批量增大仍可能再次触发问题，且会增加Kafka集群的存储和传输压力。
• 优化批量处理逻辑：在抛出BatchListenerFailedException前，先将整批消息拆分为单条，只传递失败的那条消息到异常处理流程，但这种方式会增加代码复杂度，需要结合业务场景评估。

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