GH-3424: Customizing DeadLetterPublishingRecovererFactory logging (#3461)

* GH-3424: Customizing DeadLetterPublishingRecovererFactory logging

Fixes: #3424

Provide the ability to customize logging in `DeadLetterPublishingRecovererFactory`.

* PR review addressing

Adding whats-new

Author: sobychacko

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -40,3 +40,8 @@ For more details, see xref:kafka/events.adoc[Application Events] and `Concurrent
 
 When using `ReplyingKafkaTemplate`, if the original record from the request contains a key, then that same key will be part of the reply as well.
 For more details, see xref:kafka/sending-messages.adoc[Sending Messages] section of the reference docs.
+
+[[x33-customize-logging-in-DeadLetterPublishingRecovererFactory]]
+=== Customizing Logging in DeadLetterPublishingRecovererFactory
+
+When using `DeadLetterPublishingRecovererFactory`, the user applications can override the `maybeLogListenerException` method to customize the logging behavior.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2023 the original author or authors.
+ * Copyright 2018-2024 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.
@@ -54,6 +54,7 @@
  *
  * @author Tomaz Fernandes
  * @author Gary Russell
+ * @author Soby Chacko
  * @since 2.7
  *
  */
@@ -237,6 +238,35 @@ public DeadLetterPublishingRecoverer create(String mainListenerId) {
 		return recoverer;
 	}
 
+	/**
+	 * Log the exception before sending the record in error to the retry topic.
+	 * This method can be overridden by downstream applications to customize how the error is logged.
+	 * @param exception the exception that caused the error
+	 * @param consumerRecord the original consumer record
+	 * @param nextDestination the next topic where the record goes
+	 * @since 3.3.0
+	 */
+	protected void maybeLogListenerException(Exception exception, ConsumerRecord<?, ?> consumerRecord, DestinationTopic nextDestination) {
+		if (nextDestination.isDltTopic()
+				&& !ListenerExceptionLoggingStrategy.NEVER.equals(this.loggingStrategy)) {
+			LOGGER.error(exception, () -> getErrorMessage(consumerRecord) + " and won't be retried. "
+					+ "Sending to DLT with name " + nextDestination.getDestinationName() + ".");
+		}
+		else if (nextDestination.isNoOpsTopic()
+				&& !ListenerExceptionLoggingStrategy.NEVER.equals(this.loggingStrategy)) {
+			LOGGER.error(exception, () -> getErrorMessage(consumerRecord) + " and won't be retried. "
+					+ "No further action will be taken with this record.");
+		}
+		else if (ListenerExceptionLoggingStrategy.EACH_ATTEMPT.equals(this.loggingStrategy)) {
+			LOGGER.error(exception, () -> getErrorMessage(consumerRecord) + ". "
+					+ "Sending to retry topic " + nextDestination.getDestinationName() + ".");
+		}
+		else {
+			LOGGER.debug(exception, () -> getErrorMessage(consumerRecord) + ". "
+					+ "Sending to retry topic " + nextDestination.getDestinationName() + ".");
+		}
+	}
+
 	private DeadLetterPublishingRecoverer create(
 			Function<ProducerRecord<?, ?>, KafkaOperations<?, ?>> templateResolver,
 			BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver) {
@@ -271,27 +301,6 @@ private DeadLetterPublishingRecoverer create(
 		};
 	}
 
-	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();
 	}