Spring cloud Stream Kafka DLQ throw RecordTooLargeException for Batch mode

Question

I am using Spring cloud Stream kafka Binder in my project. We have enabled the batch-mode true and batch size is 500 messages. We are using DeadLetterPublishingRecoverer to send the message to DLQ. If any message failed then we throw BatchListenerFailedException. This functionality worked fine and messages are sending into DLQ.

Recently we have an issue where while sending the message to DLQ, Kafka throws "RecordTooLargeException: Message is 15762601 bytes when serialized which is larger than 1048576". Our individual message size is in bytes. What I observed is Kafka read the messages in batch which is 500 messages and if any 1 message failed, Kafka wrap all the 500 messages with exceptions and put it into Kafka Header in "kafka_dlt-exception-message" and "kafka_dlt-exception-stacktrace". And due this message size grows further.

I can remove those 2 headers("kafka_dlt-exception-message" and "kafka_dlt-exception-stacktrace") to get rid of this issue but my question is, Is this a right way to handle this issue?

Below is the error stack trace I am getting. I am using Spring cloud stream version 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)

Below is consumer function

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

After debugging the issue I Observed that method AbstractMessageHandler -> handleMessage inside the exception block, code is passing all the messages to MessageHandlingException class which I thought it should be passing only failed message(I may be wrong) and due to that the header "kafka_dlt-exception-message" and "kafka_dlt_exception-stacktrace" contains all the messages and the message size grows.

Thanks

Spring cloud Stream Kafka DLQ throw RecordTooLargeException for Batch mode

Answers (1)

Related Questions