GH-2212: Log listener exception in retry topic flow (#2213)

* GH-2212: Log listener ex. in retry topic flow

Resolves #2212

As in Retry Topic flow recovery is successful, the exception thrown by the listener was never being logged.

Now it logs at DEBUG level for intermediate retries and ERROR level when retries are exhausted.

Also some polishing in TimestampedException as it was cluttering the stacktrace with duplicated information from cause.getMessage(). Now it logs the time when the exception occurred.

* Address TimestampedException review suggestions

Add ListenerExceptionLoggingStrategy

Add unit tests

* Address review comments

Author: tomazfernandes

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -34,14 +34,41 @@ public class TimestampedException extends KafkaException {
 
 	private final long timestamp;
 
+	@Deprecated
 	public TimestampedException(Exception ex, Clock clock) {
 		super(ex.getMessage(), ex);
 		this.timestamp = Instant.now(clock).toEpochMilli();
 	}
 
+	/**
+	 * Constructs an instance with the provided cause
+	 * and the current time.
+	 * @param ex the exception cause.
+	 */
 	public TimestampedException(Exception ex) {
-		super(ex.getMessage(), ex);
-		this.timestamp = Instant.now(Clock.systemDefaultZone()).toEpochMilli();
+		this(ex, Instant.now());
+	}
+
+	/**
+	 * Creates an instance with the timestamp of when it was thrown and its cause.
+	 * @param ex the exception cause.
+	 * @param timestamp the millis from epoch of when the exception was thrown.
+	 * @since 2.7.13
+	 */
+	public TimestampedException(Exception ex, long timestamp) {
+		super("Exception thrown at " + Instant.ofEpochMilli(timestamp), ex);
+		this.timestamp = timestamp;
+	}
+
+	/**
+	 * Creates an instance with the Instant of when it was thrown and its cause.
+	 * @param ex the exception cause.
+	 * @param now the Instant of when the exception was thrown.
+	 * @since 2.7.13
+	 */
+	public TimestampedException(Exception ex, Instant now) {
+		super("Exception thrown at " + now, ex);
+		this.timestamp = now.toEpochMilli();
 	}
 
 	public long getTimestamp() {

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@
 
 import java.math.BigInteger;
 import java.time.Clock;
+import java.time.Instant;
 import java.util.Optional;
 
 import org.apache.kafka.clients.consumer.Consumer;
@@ -96,7 +97,7 @@ public void onMessage(ConsumerRecord<K, V> consumerRecord, @Nullable Acknowledgm
 			invokeDelegateOnMessage(consumerRecord, acknowledgment, consumer);
 		}
 		catch (Exception ex) {
-			throw new TimestampedException(ex, this.clock);
+			throw new TimestampedException(ex, Instant.now(this.clock));
 		}
 	}
 

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

@@ -54,7 +54,8 @@
  */
 public class DeadLetterPublishingRecovererFactory {
 
-	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(DeadLetterPublishingRecovererFactory.class));
+	private static final LogAccessor LOGGER =
+			new LogAccessor(LogFactory.getLog(DeadLetterPublishingRecovererFactory.class));
 
 	private final DestinationTopicResolver destinationTopicResolver;
 
@@ -66,6 +67,8 @@ public class DeadLetterPublishingRecovererFactory {
 
 	private BiFunction<ConsumerRecord<?, ?>, Exception, Headers> headersFunction;
 
+	private ListenerExceptionLoggingStrategy loggingStrategy = ListenerExceptionLoggingStrategy.AFTER_RETRIES_EXHAUSTED;
+
 	public DeadLetterPublishingRecovererFactory(DestinationTopicResolver destinationTopicResolver) {
 		this.destinationTopicResolver = destinationTopicResolver;
 	}
@@ -121,6 +124,24 @@ public boolean removeNotRetryableException(Class<? extends Exception> exceptionT
 		return this.nonFatalExceptions.add(exceptionType);
 	}
 
+	/**
+	 * Never logs the listener exception.
+	 * The default is logging only after retries are exhausted.
+	 * @since 2.7.13
+	 */
+	public void neverLogListenerException() {
+		this.loggingStrategy = ListenerExceptionLoggingStrategy.NEVER;
+	}
+
+	/**
+	 * Logs the listener exception at each attempt.
+	 * The default is logging only after retries are exhausted.
+	 * @since 2.7.13
+	 */
+	public void alwaysLogListenerException() {
+		this.loggingStrategy = ListenerExceptionLoggingStrategy.EACH_ATTEMPT;
+	}
+
 	@SuppressWarnings("unchecked")
 	public DeadLetterPublishingRecoverer create() {
 		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(// NOSONAR anon. class size
@@ -185,11 +206,45 @@ private TopicPartition resolveDestination(ConsumerRecord<?, ?> cr, Exception e)
 				? "none"
 				: nextDestination.getDestinationName()));
 
+		maybeLogListenerException(e, cr, nextDestination);
+
 		return nextDestination.isNoOpsTopic()
 					? null
 					: resolveTopicPartition(cr, nextDestination);
 	}
 
+	private void maybeLogListenerException(Exception e, ConsumerRecord<?, ?> cr, DestinationTopic nextDestination) {
+		if (nextDestination.isDltTopic()
+				&& !ListenerExceptionLoggingStrategy.NEVER.equals(this.loggingStrategy)) {
+			LOGGER.error(e, () -> getErrorMessage(cr) + " and won't be retried. "
+					+ "Sending to DLT with name " + nextDestination.getDestinationName() + ".");
+		}
+		else if (nextDestination.isNoOpsTopic()
+				&& !ListenerExceptionLoggingStrategy.NEVER.equals(this.loggingStrategy)) {
+			LOGGER.error(e, () -> getErrorMessage(cr) + " and won't be retried. "
+					+ "No further action will be taken with this record.");
+		}
+		else if (ListenerExceptionLoggingStrategy.EACH_ATTEMPT.equals(this.loggingStrategy)) {
+			LOGGER.error(e, () -> getErrorMessage(cr) + ". "
+					+ "Sending to retry topic " + nextDestination.getDestinationName() + ".");
+		}
+		else {
+			LOGGER.debug(e, () -> getErrorMessage(cr) + ". "
+					+ "Sending to retry topic " + nextDestination.getDestinationName() + ".");
+		}
+	}
+
+	private static String getErrorMessage(ConsumerRecord<?, ?> cr) {
+		return "Record: " + getRecordInfo(cr) + " threw an error at topic " + cr.topic();
+	}
+
+	private static String getRecordInfo(ConsumerRecord<?, ?> cr) {
+		Header originalTopicHeader = cr.headers().lastHeader(KafkaHeaders.ORIGINAL_TOPIC);
+		return String.format("topic = %s, partition = %s, offset = %s, main topic = %s",
+				cr.topic(), cr.partition(), cr.offset(),
+				originalTopicHeader != null ? new String(originalTopicHeader.value()) : cr.topic());
+	}
+
 	/**
 	 * Creates and returns the {@link TopicPartition}, where the original record should be forwarded.
 	 * By default, it will use the partition same as original record's partition, in the next destination topic.
@@ -284,6 +339,23 @@ private Header getOriginaTimeStampHeader(ConsumerRecord<?, ?> consumerRecord) {
 		return consumerRecord.headers()
 					.lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP);
 	}
-}
 
+	private enum ListenerExceptionLoggingStrategy {
+
+		/**
+		 * Never log the listener exception.
+		 */
+		NEVER,
+
+		/**
+		 * Log the listener exception after each attempt.
+		 */
+		EACH_ATTEMPT,
 
+		/**
+		 * Log the listener only after retries are exhausted.
+		 */
+		AFTER_RETRIES_EXHAUSTED
+
+	}
+}