RetryableTopic - breaking changes in the error handler

[v-chernyshev]

In what version(s) of Spring for Apache Kafka are you seeing this issue?

2.8.4.

Describe the bug

ListenerContainerFactoryConfigurer now creates its error handler here:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java

Lines 235 to 246 in 93efeb4

	protected CommonErrorHandler createErrorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer,
	Configuration configuration) {
	DefaultErrorHandler errorHandler = createDefaultErrorHandlerInstance(deadLetterPublishingRecoverer);
	errorHandler.defaultFalse();
	errorHandler.setCommitRecovered(true);
	errorHandler.setLogLevel(KafkaException.Level.DEBUG);
	if (this.blockingExceptionTypes != null) {
	errorHandler.addRetryableExceptions(this.blockingExceptionTypes);
	}
	this.errorHandlerCustomizer.accept(errorHandler);
	return errorHandler;
	}

There are two breaking changes/regressions compared to version 2.8.3:

• createDefaultErrorHandlerInstance no longer provides a "no retry" back off strategy by default. The outcome of this change is that a to-be-recovered record is retried 10 times instead of immediately being sent into the next recovery topic.
• errorHandler.defaultFalse completely disables the default classification of the fatal exceptions (such as DeserializationException.class). In fact, every classification request now returns false.

The second change causes every back off exception to be printed at error level due to the code here:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java

Lines 143 to 164 in f0ad7b0

	protected RecoveryStrategy getRecoveryStrategy(List<ConsumerRecord<?, ?>> records,
	@Nullable Consumer<?, ?> recoveryConsumer, Exception thrownException) {
	if (getClassifier().classify(thrownException)) {
	return this.failureTracker::recovered;
	}
	else {
	try {
	this.failureTracker.getRecoverer().accept(records.get(0), recoveryConsumer, thrownException);
	this.failureTracker.getRetryListeners().forEach(rl -> rl.recovered(records.get(0), thrownException));
	}
	catch (Exception ex) {
	if (records.size() > 0) {
	this.logger.error(ex, () -> "Recovery of record ("
	+ KafkaUtils.format(records.get(0)) + ") failed");
	this.failureTracker.getRetryListeners().forEach(rl ->
	rl.recoveryFailed(records.get(0), thrownException, ex));
	}
	return (rec, excep, cont, consumer) -> NEVER_SKIP_PREDICATE.test(rec, excep);
	}
	return (rec, excep, cont, consumer) -> ALWAYS_SKIP_PREDICATE.test(rec, excep);
	}
	}

getClassifier().classify(thrownException) now always returns false and this.failureTracker.getRecoverer().accept may throw a back off exception, which does not reach the following special back off exception handler any longer:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekUtils.java

Lines 109 to 113 in 93efeb4

	catch (Exception ex) {
	if (isBackoffException(ex)) {
	logger.debug(ex, () -> ListenerUtils.recordToString(record)
	+ " included in seeks due to retry back off");
	}

To Reproduce

Unfortunately, I do not have a sample right now, but I will try to prepare it this week. I believe that any application that uses the retryable topics recovery mechanism is impacted.

Expected behavior

I expect the default behaviour of the retryable topics recovery system to be identical to version 2.8.3.

[v-chernyshev]

A slight clarification regarding the back off exception handler changes. The passed thrownException may be a back off exception, but then it is unwrapped here:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java

Lines 176 to 179 in 93efeb4

	private TopicPartition resolveDestination(ConsumerRecord<?, ?> cr, Exception e) {
	if (SeekUtils.isBackoffException(e)) {
	throw (NestedRuntimeException) e; // Necessary to not commit the offset and seek to current again
	}

The catch statement of getRecoveryStrategy intercepts it and prints an error.

[tomazfernandes]

Hi @v-chernyshev, thanks for reporting this.

There are two breaking changes/regressions compared to version 2.8.3:

• createDefaultErrorHandlerInstance no longer provides a "no retry" back off strategy by default. The outcome of this change is that a to-be-recovered record is retried 10 times instead of immediately being sent into the next recovery topic.
• errorHandler.defaultFalse completely disables the default classification of the fatal exceptions (such as DeserializationException.class). In fact, every classification request now returns false.

I don't see this behavior in my tests. Since all classifications now return false by default, they skip the 10 retrials, and only attempt retrials if you add exceptions to be retried on the LCFC. Any further details you can share on that would be appreciated.

The second change causes every back off exception to be printed at error level due to the code here:

This one I can reproduce. The idea is that KafkaBackOffExceptions should be logged at it's own level, and we now set it to DEBUG. I'll try to come with a workaround for this while we look for the fix.

[tomazfernandes]

Also, the previous FATAL classifications in DLPR were in practice no-ops - since the policy was no retries, it didn't matter whether the exception was FATAL or not. We do have the same FATAL classifications now on the DefaultDestinationTopicResolver class and those exceptions are now forwarded directly to the DLT by default. You can add and remove exceptions from the list.

[tomazfernandes]

A slight clarification regarding the back off exception handler changes. The passed thrownException may be a back off exception, but then it is unwrapped here:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java

Lines 176 to 179 in 93efeb4

	private TopicPartition resolveDestination(ConsumerRecord<?, ?> cr, Exception e) {
	if (SeekUtils.isBackoffException(e)) {
	throw (NestedRuntimeException) e; // Necessary to not commit the offset and seek to current again
	}

The catch statement of getRecoveryStrategy intercepts it and prints an error.

Just for clarification, the exception isn't unwrapped here, it's just thrown again.

[v-chernyshev]

Just for clarification, the exception isn't unwrapped here, it's just thrown again.

Yes, you are absolutely right, I didn't phrase myself properly.

I don't see this behavior in my tests. Since all classifications now return false by default

All right, I'll try to come up with an example based on our code, hopefully by the end of tomorrow. Thanks for looking into it!

[tomazfernandes]

@v-chernyshev ok, thanks. I'll keep looking for the behavior you described.

Can you just please check if by any chance you're adding exceptions to the DeadLetterPublishingRecoverer classification? That would be the most likely cause.

[v-chernyshev]

Can you just please check if by any chance you're adding exceptions to the DeadLetterPublishingRecoverer classification? That would be the most likely cause.

Yes, your guess is correct. In my tests I added all the default exceptions back via setClassifications and replaced the default value with true to partially restore the 2.8.3 behaviour. The now default back off strategy, however, has this.noRetries in FailedRecordTracker.recovered set to false, so the else branch is picked. Then nextBackOff is equal to zero, the thread sleeps for 10 milliseconds in ListenerUtils.stoppableSleep and then false is returned from the method.

[tomazfernandes]

We have this logic in FailedRecordProcessor.getRecoveryStrategy:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java

Lines 145 to 152 in 1d13671

	if (getClassifier().classify(thrownException)) {
	return this.failureTracker::recovered;
	}
	else {
	try {
	this.failureTracker.getRecoverer().accept(records.get(0), recoveryConsumer, thrownException);
	this.failureTracker.getRetryListeners().forEach(rl -> rl.recovered(records.get(0), thrownException));
	}

Since the classification returns false, it should go to the else branch which has no sleep, which is what happens on my tests.

Looks right?

[v-chernyshev]

Yes, with an empty classifier it indeed always goes into the else branch. Unfortunately, the only way that I could find to avoid getting the back off exception logs at error level with version 2.8.4 was to redirect the code flow to return this.failureTracker::recovered :(

Admittedly, I do not really know how the return value of getClassifier().classify impacts the various parts of Spring Kafka. I'll review the 2.8.4 changelog to better understand the exception classifier changes.

[tomazfernandes]

Ok, so just to make sure we're on the same page. Functionality-wise the framework is behaving as expected - it only retries when it should.

The problem at hand is that KafkaBackOffExceptions are being logged as ERROR - and that's what caught your attention in the first place.

Does that sound about right?

[tomazfernandes]

We've documented the changes here

[v-chernyshev]

Thanks for the link, somehow I've missed this section. And yes, all of it started with the back off exceptions showing up in the logs, which is the only issue. Everything else is due to my misunderstanding of the changes, sorry for the confusion.