RetryableTopic system breaks message on deserialization errors causes it to grow unboundedly

[mh-dev]

🐞 Bug report
Version: 2.7.8

We did run into an issue that a message did grow unboundedly on a deserialization issue when using the RetryableTopic system with a DltHandler. After hours of troubleshooting and code deep dive we identified the issue in how the message data gets re-published in that case. What is happening is that the original message gets stored as byte array in DeserializationException.java#L40. This data block gets then forwarded to the KafkaTemplate which is then using the JsonSerializer. This serializer is internally using writer.writeValueAsBytes(data).
The behavior of this is that the data grows and for every pass through this continues to grow further and end up in a endless spiral with RecordTooLargeException.

See the following repo to force the message corruption.

Edit replaced the example with a repo that reproduces the issue

[garyrussell]

A DeserializationException will not be created with that configuration - it needs an ErrorHandlingDeserializer.

Even then, I don't see how that byte[] would increase on each pass because it just contains the original value that failed to be deserialized.

Perhaps you are experiencing the problem found (and fixed) last week?

https://docs.spring.io/spring-kafka/docs/2.7.9-SNAPSHOT/reference/html/#dlpr-headers

It is in 2.7.9-SNAPSHOT and 2.7.9 will be released shortly.

If that's not it, please provide a minimal, complete, verifiable example that exhibits the behavior you are seeing.

[mh-dev]

The example code is just a minimal setup that run into similar size errors. I can't guarantee adhoc that creates the exact same issue.
We have been using an ErrorHandlingDeserializer in the real life example. It was not related to the headers, since we have been removing them already because size issues. We also confirmed this by debug code.
The most severe aspect of the issue itself is that the message got corrupted. Technically we found a way to recover it by reversing the incorrect serialization multiple times.

[garyrussell]

The example code is just a minimal setup that run into similar size errors.

So please provide the full project so we can take a look to see what's happening.

The most severe aspect of the issue itself is that the message got corrupted. Technically we found a way to recover it by reversing the incorrect serialization multiple times.

That makes no sense to me; again, a complete example would be helpful.

[mh-dev]

Will setup a new repo with a more complete example. This might take 1-2 days.

[mh-dev]

Sorry for the delay see the following repo for an example that causes the message to get larger on every retry. Details see in its readme.
I also updated the original message and replaced the example there too.

[garyrussell]

The issue is that for deserialization exceptions, we don't want to use the JsonSerializer since we already have a byte[].

In 2.8, we added a new DelegatingByTypeSerializer -

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTypeSerializer.java

Lines 29 to 36 in df4a5b1

	/**
	* Delegates to a serializer based on type.
	*
	* @author Gary Russell
	* @since 2.8
	*
	*/
	public class DelegatingByTypeSerializer implements Serializer<Object> {

I copied that serializer into your project and reconfigured thus:

  @Bean
  public RetryTopicConfiguration retryableTopic(
      ConcurrentKafkaListenerContainerFactory<String, HelloMessage> helloListenerContainerFactory,
      KafkaTemplate<String, Object> kafkaTemplate) {

    return RetryTopicConfigurationBuilder
        .newInstance()
        .maxAttempts(10)
        .includeTopic("hello-topic")
        .retryTopicSuffix(String.format(RETRY_TOPIC_SUFFIX, GROUP_ID))
        .doNotRetryOnDltFailure()
        .dltSuffix(String.format(DLT_TOPIC_SUFFIX, GROUP_ID))
        .dltHandlerMethod(HelloListener.class, "dltHello")
        .useSingleTopicForFixedDelays()
        .doNotAutoCreateRetryTopics()
        .listenerFactory(helloListenerContainerFactory)
        .create(kafkaTemplate);
  }

...

 @Bean
  public ProducerFactory<String, Object> helloProducerFactory() {
    return new DefaultKafkaProducerFactory<>(getHelloProducerConfiguration(), new StringSerializer(),
      new DelegatingByTypeSerializer(Map.of(byte[].class, new ByteArraySerializer(),
            HelloMessage.class, new JsonSerializer<Object>())));
  }

  @Bean
  public KafkaTemplate<String, Object> helloKafkaTemplate() {
    return new KafkaTemplate<>(helloProducerFactory());
  }

(template is now <String, Object>).

2021-11-09 11:58:50.888  INFO 58515 --- [ntainer#0-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : Message value type: class [B
2021-11-09 11:58:50.888  INFO 58515 --- [ntainer#0-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : byte value length 4
2021-11-09 11:58:50.890  INFO 58515 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-1] Cluster ID: saMBGtXQTdaX-E4Kqr6PFQ
2021-11-09 11:58:50.904  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : Message value type: class [B
2021-11-09 11:58:50.905  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : byte value length 4
2021-11-09 11:58:50.911  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : Message value type: class [B
2021-11-09 11:58:50.911  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : byte value length 4
2021-11-09 11:58:50.917  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : Message value type: class [B
2021-11-09 11:58:50.917  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : byte value length 4
2021-11-09 11:58:50.922  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : Message value type: class [B
2021-11-09 11:58:50.922  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : byte value length 4
2021-11-09 11:58:50.928  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : Message value type: class [B
2021-11-09 11:58:50.928  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : byte value length 4
2021-11-09 11:58:50.933  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : Message value type: class [B
2021-11-09 11:58:50.933  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : byte value length 4
2021-11-09 11:58:50.939  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : Message value type: class [B
2021-11-09 11:58:50.939  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : byte value length 4
2021-11-09 11:58:50.948  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : Message value type: class [B
2021-11-09 11:58:50.948  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : byte value length 4
2021-11-09 11:58:50.959  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : Message value type: class [B
2021-11-09 11:58:50.959  INFO 58515 --- [p-retries-0-C-1] .e.c.e.c.ValueLoggingProducerInterceptor : byte value length 4