From fe95ac4918bef798c7e7766c6456e6eff62d304f Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 17 Feb 2025 14:03:30 -0600 Subject: [PATCH] Use separate logger for PulsarListener exceptions (#1039) This creates a separate logger for DefaultPulsarMessageListenerContainer that it uses to log exceptions thrown from listener callback methods. The exceptions are still logged at debug level in order to not change behavior in a patch release. However, the log category used by the logger can then be set to debug level but not spam the logs with the other debug statements in the listener container. Also, adds exception logging to the batch listener invocation using the same listener error logger as the record listener invocation. Resolves #1008 --- .../DefaultPulsarMessageListenerContainer.java | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java index baba3e84..6f28f064 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java @@ -111,6 +111,9 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess private final Condition pausedCondition = this.lockOnPause.newCondition(); + private final LogAccessor listenerErrorLogger = new LogAccessor( + "%s-ListenerErrors".formatted(DefaultPulsarMessageListenerContainer.class.getName())); + public DefaultPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProperties) { super(pulsarConsumerFactory, pulsarContainerProperties); @@ -629,7 +632,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess inRetryMode.compareAndSet(true, false); } catch (RuntimeException e) { - DefaultPulsarMessageListenerContainer.this.logger.debug(e, + DefaultPulsarMessageListenerContainer.this.listenerErrorLogger.debug(e, () -> "Error dispatching the message to the listener."); if (this.pulsarConsumerErrorHandler != null) { invokeRecordListenerErrorHandler(inRetryMode, message, e, txn); @@ -642,9 +645,9 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess this.nackableMessages.add(message.getMessageId()); } else { - throw new IllegalStateException("Exception occurred and message %s was not auto-nacked; " - + "switch to AckMode BATCH or RECORD to enable auto-nacks" - .formatted(message.getMessageId()), + throw new IllegalStateException( + "Exception occurred and message %".formatted(message.getMessageId()) + + "was not auto-nacked; switch to AckMode BATCH or RECORD to enable auto-nacks", e); } } @@ -713,6 +716,8 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess return Collections.emptyList(); } catch (RuntimeException ex) { + DefaultPulsarMessageListenerContainer.this.listenerErrorLogger.debug(ex, + () -> "Error dispatching the messages to the batch listener."); if (this.pulsarConsumerErrorHandler != null) { return invokeBatchListenerErrorHandler(inRetryMode, messagesPendingInBatch, messageList, ex, txn); }