Logging of next retry in RetryTopicConfiguration - Spring kafka retry

[hariraogotit]

Team,
I am using spring kafka retry - RetryTopicConfiguration. A question, any lead is much appreciated please.

Is there a way to extract when will be the next retry?

eg say we have 2 retry topics topic-2000, topic-6000

We know that when it writes to topic-2000, it will process after 2000ms and similar when it writes to 6000, it will process after 6000ms but is there a way to extract information about how long it will take for retry to kick in?

I want to log something like retry will happen after 2000ms. If it is not successful then I want to log that retry will happen in 6000ms.

[hariraogotit]

Team, I am using spring kafka retry - RetryTopicConfiguration. A question, any lead is much appreciated please.

Is there a way to extract when will be the next retry?

eg say we have 2 retry topics topic-2000, topic-6000

We know that when it writes to topic-2000, it will process after 2000ms and similar when it writes to 6000, it will process after 6000ms but is there a way to extract information about how long it will take for retry to kick in?

I want to log something like retry will happen after 2000ms. If it is not successful then I want to log that retry will happen in 6000ms.

i see the line "if a message processing throws an exception, the configured DefaultErrorHandler and DeadLetterPublishingRecoverer forwards the message to the next topic, using a DestinationTopicResolver to know the next topic and the delay for it."

in https://docs.spring.io/spring-kafka/api/org/springframework/kafka/retrytopic/RetryTopicConfigurer.html

And I see List<DestinationTopic.Properties> destinationTopicProperties; in RetryTopicConfiguration

But when I follow along I am not able to get hold of DestinationTopicResolver.

[artembilan]

The KafkaBackoffAwareMessageListenerAdapter has a logic like this:

		maybeGetBackoffTimestamp(consumerRecord)
				.ifPresent(nextExecutionTimestamp -> this.kafkaConsumerBackoffManager
						.backOffIfNecessary(createContext(consumerRecord, nextExecutionTimestamp, consumer)));

and that one does this:

	public void backOffIfNecessary(Context context) {
		long backoffTime = context.getDueTimestamp() - System.currentTimeMillis();
		LOGGER.debug(() -> "Back off time: " + backoffTime + " Context: " + context);
		if (backoffTime > 0) {
			pauseConsumptionAndThrow(context, backoffTime);
		}
	}

Isn't that debug message not enough for you?

[hariraogotit]

In all of my KafkaListeners I am planning to extract from the headers as below

Header ot = consumerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP);
Header count = consumerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
Header bt = consumerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP);

[artembilan]

That's no OK. The record will reach your listener only after back off.
So, even if you receive the info eventually, it does not give you a desired effect you have talked in the beginning for determining when the next retry after delay.

[hariraogotit]

I tried this in a play ground project with one kafka listener.

When I send a proper message the headers are all null but when I send a poison pill (bad data - hari sydney). The headers ot, count and bt are not null.

Count gives the count, I am able to get the original time and back off time i,e it looks like it is the time when the next retry will kick in. Some logs below

Topic name from listener Sydney -> APZ-sydney
Countnull
okay
Message from the listener Sydney --> hari sydney

Topic name from listener Sydney -> APZ-sydney-AP-Australia-retry-1000
Original time stamp 1705597193130
2024-01-18T16:59:53.130
CountRecordHeader(key = retry_topic-attempts, value = [0, 0, 0, 2])
Back off time 1705597195153
2024-01-18T16:59:55.153
okay
Message from the listener Sydney --> hari sydney

2024-01-18T16:59:55.697Z  INFO 70197 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-1] Resetting the last seen epoch of partition APZ-sydney-AP-Australia-retry-2000-0 to 0 since the associated topicId changed from null to DSL0fzmMRouv97MUEAd6Nw
2024-01-18T16:59:55.703Z  INFO 70197 --- [etry-2000-0-C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-sydney-AP-Australia-retry-2000-2, groupId=sydney-AP-Australia-retry-2000] Seeking to offset 38 for partition APZ-sydney-AP-Australia-retry-2000-0
Topic name from listener Sydney -> APZ-sydney-AP-Australia-retry-2000
Original time stamp 1705597193130
2024-01-18T16:59:53.130
CountRecordHeader(key = retry_topic-attempts, value = [0, 0, 0, 3])
Back off time 1705597197196
2024-01-18T16:59:57.196
okay
Message from the listener Sydney --> hari sydney

2024-01-18T16:59:57.709Z  INFO 70197 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-1] Resetting the last seen epoch of partition APZ-sydney-AP-Australia-retry-4000-0 to 0 since the associated topicId changed from null to YvYG68RZSjC-YSddFX4lOg
2024-01-18T16:59:57.713Z  INFO 70197 --- [etry-4000-0-C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-sydney-AP-Australia-retry-4000-4, groupId=sydney-AP-Australia-retry-4000] Seeking to offset 39 for partition APZ-sydney-AP-Australia-retry-4000-0
Topic name from listener Sydney -> APZ-sydney-AP-Australia-retry-4000
Original time stamp 1705597193130
2024-01-18T16:59:53.130
CountRecordHeader(key = retry_topic-attempts, value = [0, 0, 0, 4])
Back off time 1705597201202
2024-01-18T17:00:01.202
okay
Message from the listener Sydney --> hari sydney


2024-01-18T17:00:01.209Z ERROR 70197 --- [etry-4000-0-C-1] k.r.DeadLetterPublishingRecovererFactory : Record: topic = APZ-sydney-AP-Australia-retry-4000, partition = 0, offset = 39, main topic = APZ-sydney threw an error at topic APZ-sydney-AP-Australia-retry-4000 and won't be retried. Sending to DLT with name APZ-sydney-AP-Australia-dlt.

I converted the long timestamp to a readable one.
Eg the back off 2024-01-18T16:59:55.153 is the next retry time as you can see the next retry is attempted as shown below on the specified time
2024-01-18T16:59:55.697Z INFO 70197 --- [ad | producer-1] org.apache.kafka.clients.Metadata

[artembilan]

OK. It is still hard for me to understand what is going on, but sounds like we can close this discussion since you got the answer for your original request.

[hariraogotit]

Apologies I formatted it now

basically the below i.e back off time gives when the next retry will kick in and next retry happens at that time

Back off time 1705597197196
2024-01-18T16:59:57.196

2024-01-18T16:59:57.709Z INFO 70197 --- [ad | producer-1] org.apache.kafka.clients.Metadat