DeserializationException recovery with non-blocking retries

[jgslima]

When using non-blocking retries, and listeners that receive a single record, is there some way to use ErrorHandlingDeserializer in a way that the application code has means to act on (ie to recover) records with deserialization issues?

When not using non-blocking retries, it is clear that the configured ConsumerRecordRecoverer is called so that the application can decide what to do with the original byte[]s.

When using non-blocking retries (and having prepared the KafkaTemplate used to publish in the retry and DLT topics with a DelegatingByTypeSerializer, in order to handle byte[]), I can see that the record is published in the DLT topic. However, the DTL handler method is not called. In this point, it seems that the listener container of the DLT works like any other listener: before actually invoking the listener, the container checks for DeserializationException in the headers.

Also, as the DLT is the last in the chain, there is no another level of record recoverer. What happens is just a log with the message "Record: topic = NON_BLOCKING_RETRIES-dlt, partition = 0, offset = 0, main topic = NON_BLOCKING_RETRIES threw an error at topic NON_BLOCKING_RETRIES-dlt and won't be retried. No further action will be taken with this record".

I need to have access to, and decide what to do with these records. Some applications of ours here cannot afford to "lose" records.

A possibility might be to configure a plain ConsumerRecordRecoverer to be called in errors of the DLT listener. Not sure if this would be the best approach.

[jgslima]

Well, I figured out a way to achieve what I want.

To declare an explicit @KafkaListener associated to all the DLTs of the topics that have non blocking retries. In fact, this is a dummy listener (its method is actually empty), because records that do not have DeserializationException will be handled by the internal listener container provided by the framework, which will call the official DLT handling method.

But this listener is not associated with non blocking retries. This way, I can configure in the regular DefaultErrorHandler a custom record recoverer. Effectively, I have a record recoverer for the DLT topics, and in there I can deal with ConsumerRecords with DeserializationException in the headers.

[jgslima]

I will wait if there is another suggestion, before marking my own answer as the accepted one.

[garyrussell]

Sorry; I had this answer staged and I was working on option 3 but had to run out for a couple of hours.

Your solution is simpler, but I will complete my example and post it here.

Interesting point.

The container knows nothing about non-blocking retries, DLTs, etc, etc.

The records are not actually "lost" they are still sitting in the DLT topic.

I can think of a few solutions

1. Create a KafkaConsumer from the DefaultKafkaConsumerFactory and use it directly with a ByteArrayDeserializer to consume from the DLT topic (instead of using a listener container).
2. Add a producer interceptor and, when publishing one of these to the DLT topic, move the value() to a header and set the value to null (and strip the deserialization headers).
3. Reconfigure the DLT handler container to use a ByteArrayDeserializer

The last one is probably the simplest.

[garyrussell]

Here is my proposed solution:

@SpringBootApplication
public class Kgh2460Application extends RetryTopicConfigurationSupport {

	public static void main(String[] args) {
		SpringApplication.run(Kgh2460Application.class, args);
	}

	@Override
	protected void configureCustomizers(CustomizersConfigurer configurer) {
		Properties props = new Properties();
		props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
		configurer.customizeListenerContainer(container -> {
			if (container.getContainerProperties().getTopics()[0].endsWith("-dlt")) {
				container.getContainerProperties().setKafkaConsumerProperties(props);
			}
		});
	}

	@Bean
	NewTopic topic() {
		return TopicBuilder.name("kgh2460").partitions(1).replicas(1).build();
	}

	@Bean
	TaskScheduler sched() {
		return new ThreadPoolTaskScheduler();
	}

	@Bean
	ApplicationRunner runner(DefaultKafkaProducerFactory<String, Object> pf) {

		KafkaTemplate<String, Object> stringTemplate = new KafkaTemplate<>(pf,
				Map.of(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class));

		Map<Class<?>, Serializer> delegates = new LinkedHashMap<>();
		delegates.put(byte[].class, new ByteArraySerializer());
		delegates.put(Object.class, new JsonSerializer<Object>());
		pf.setValueSerializer(new DelegatingByTypeSerializer(delegates));

		return args -> {
			stringTemplate.send("kgh2460", "..junk..");
		};
	}

}

@Component
class Listener {

	@RetryableTopic
	@KafkaListener(id = "kgh2460", topics = "kgh2460")
	void listen(String in) {
		System.out.println(in);
	}

	@DltHandler
	void handle(ConsumerRecord<String, byte[]> record) {
		System.out.println("DLT: " + KafkaUtils.format(record));
	}

}

DLT: kgh2460-dlt-0@19

Yours has the benefit that the DLT listener will get proper deserialization of "other" DLT records whereas this one would always get the raw byte[] regardless of the root cause exception.

[jgslima]

Feeling flattered for so many attention. :)

It is good to see you approach, because I would not know how to customize a specific container with CustomizersConfigurer. This trick of inspecting the topic name of the container may be very usefull.

I think I will go with your solution.

PS.: as you can see in the last days, I've been studying the non-blocking retries features a lot. Maybe I can share some other questions, ideas and small bugs I have here.

[garyrussell]

Thanks for the feedback; we are getting close to GA (Nov 21); our second release candidate is due Nov 3; while the feature has been around for a year and a half, we have now removed the "experimental" designation on the API for 3.0 (after making many changes in 2.9) so we don't have much time left if more API changes are needed.

Bugs are a different issue, of course.

[jgslima]

A small bug I noticed here is about the value of the Header "kafka_deliveryAttempt" (KafkaHeaders.DELIVERY_ATTEMPT).
When used in combination with non-blocking retries, from the 2nd non-blocking attempt, it starts with the value 2 instead of 1.

Example: in a configuration with 2 blocking retries (Backoff with 2 maxAttempts) and a RetryTopicConfiguration with .maxAttempts(3), one would expect the value of the Headers in each listener invocation to be:

kafka_deliveryAttempt=1, retry_topic-attempts=null
kafka_deliveryAttempt=2, retry_topic-attempts=null
kafka_deliveryAttempt=3, retry_topic-attempts=null

kafka_deliveryAttempt=1, retry_topic-attempts=2
kafka_deliveryAttempt=2, retry_topic-attempts=2
kafka_deliveryAttempt=3, retry_topic-attempts=2

kafka_deliveryAttempt=1, retry_topic-attempts=3
kafka_deliveryAttempt=2, retry_topic-attempts=3
kafka_deliveryAttempt=3, retry_topic-attempts=3

But instead we have:

kafka_deliveryAttempt=1, retry_topic-attempts=null
kafka_deliveryAttempt=2, retry_topic-attempts=null
kafka_deliveryAttempt=3, retry_topic-attempts=null

kafka_deliveryAttempt=2, retry_topic-attempts=2
kafka_deliveryAttempt=2, retry_topic-attempts=2
kafka_deliveryAttempt=3, retry_topic-attempts=2

kafka_deliveryAttempt=2, retry_topic-attempts=3
kafka_deliveryAttempt=2, retry_topic-attempts=3
kafka_deliveryAttempt=3, retry_topic-attempts=3

[garyrussell]

Please open a bug issue for that; it certainly doesn't look right.

[jgslima]

Okay