Original exception never logged with Retryable Topics

[tomazfernandes]

This SO user pointed out that, when the record goes through the Retry Topic flow, the original exception never gets logged.

We had already discussed this a bit earlier this year.

@garyrussell proposed:

But I see your point; since we always recover these failures, they are never logged. Maybe the DLPR needs an option to log the exception after successfully recovering a record (but that doesn't need to be part of this fix).

I suggested as an alternative:

We could log the error message just before we either send the record to the DLT or give up processing - that way nothing gets logged if any of the retrials work, and gets logged only once if none.

The main difference seems to be whether we want to log the exception at every retry attempt or only before it goes to the DLT.

We might look into adding both and making these configurable - users might be able to set a flag in DLPR to log after every attempt, and DLPRF to log before going to the DLT.

We could also have this as an enum in DLPRF, with something like OriginalExceptionLoggingStrategy.NEVER, EVERY_ATTEMPT, AFTER_RETRIES_EXHAUSTED - DLPRF would then set the appropriate flag in DLPR.

WDYT?

[tomazfernandes]

Just to illustrate, if we decide to have the option of logging only after retry exhaustion, this is where I think we might do it:

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

Lines 176 to 191 in f53d41b

	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
	}
	DestinationTopic nextDestination = this.destinationTopicResolver.resolveDestinationTopic(
	cr.topic(), getAttempts(cr), e, getOriginalTimestampHeaderLong(cr));
	LOGGER.debug(() -> "Resolved topic: " + (nextDestination.isNoOpsTopic()
	? "none"
	: nextDestination.getDestinationName()));
	return nextDestination.isNoOpsTopic()
	? null
	: resolveTopicPartition(cr, nextDestination);
	}

If nextDestination is dlt or noOps, we'd log the exception.

[tzie0062]

I think I like your idea of using a OriginalExceptionLoggingStrategy in the DLPR which could default to AFTER_RETRIES_EXHAUSTED, leaving users the option to handle logging within application code by setting it to NEVER.

Would it make sense to expose some metrics (e.g. failures) via micrometer as well?

[tomazfernandes]

Sure, I think the framework should be as flexible as possible to adapt to users' needs. I think we can have all logging logic handled in the DLPRF - no need to change anything in DLPR for this.

TBH, I'm not that familiar with micrometer, maybe that's a good chance to learn more about it. But it probably should be a separate issue so that one thing doesn't end up delaying the other.

I'd like to hear @garyrussell's thoughts on both the Enum and the micrometer if possible, then I can change the PR I opened with this.

Thanks for the input!

[garyrussell]

The enum solution sounds good to me.

Exceptions are already counted by all containers, with the tags including the listener id (bean name).

https://docs.spring.io/spring-kafka/docs/current/reference/html/#monitoring-listener-performance

[tzie0062]

@garyrussell just out of curiosity: if I have two retry topics and all attempts to process an event fail, what will the metrics show? One or three exceptions?

[garyrussell]

The container is not aware of the retry mechanisms; all 3 exceptions will be counted, but they will be tagged with different container ids.

[tomazfernandes]

#2212