Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3424: Customizing DeadLetterPublishingRecovererFactory logging (#3461)

Browse Source 
* GH-3424: Customizing DeadLetterPublishingRecovererFactory logging

Fixes: #3424

Provide the ability to customize logging in `DeadLetterPublishingRecovererFactory`.

* PR review addressing

Adding whats-new
This commit is contained in:
Soby Chacko
2024-08-27 11:11:57 -04:00
committed by GitHub
parent 45a8167c24
commit 29146ba759
2 changed files with 36 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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.

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

@@ -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 @@ import org.springframework.util.Assert;
*
* @author Tomaz Fernandes
* @author Gary Russell
* @author Soby Chacko
* @since 2.7
*
*/
@@ -237,6 +238,35 @@ public class DeadLetterPublishingRecovererFactory {
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 @@ public class DeadLetterPublishingRecovererFactory {
};
}
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();
}