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

Use separate logger for PulsarListener exceptions (#1039)

Browse Source 
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
This commit is contained in:
Chris Bono
2025-02-17 14:03:30 -06:00
committed by GitHub
parent 147844b07d
commit fe95ac4918
1 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
View File

@@ -111,6 +111,9 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
private final Condition pausedCondition = this.lockOnPause.newCondition();
private final LogAccessor listenerErrorLogger = new LogAccessor(
"%s-ListenerErrors".formatted(DefaultPulsarMessageListenerContainer.class.getName()));
public DefaultPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
PulsarContainerProperties pulsarContainerProperties) {
super(pulsarConsumerFactory, pulsarContainerProperties);
@@ -629,7 +632,7 @@ public class DefaultPulsarMessageListenerContainer<T> 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<T> 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<T> 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);
}