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

Polish "Fix batch payload type as List<Message<?>>"

Browse Source
This commit is contained in:
Chris Bono
2022-09-16 20:45:02 -05:00
committed by Chris Bono
parent 685472b996
commit eeb983aa31
2 changed files with 6 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -394,17 +394,14 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
*/
private List<Message<T>> invokeBatchListenerErrorHandler(AtomicBoolean inRetryMode,
AtomicBoolean messagesPendingInBatch, List<Message<T>> messageList, Throwable exception) {
try {
Assert.isInstanceOf(PulsarBatchListenerFailedException.class, exception,
"Batch listener should throw PulsarBatchListenerFailedException on errors.");
}
catch (Exception e) {
// try in the cause if something downstream wrapped the original
// exception.
// Make sure either the exception or the exception cause is batch exception
if (!(exception instanceof PulsarBatchListenerFailedException)) {
exception = exception.getCause();
Assert.isInstanceOf(PulsarBatchListenerFailedException.class, exception,
"Batch listener should throw PulsarBatchListenerFailedException on errors.");
}
PulsarBatchListenerFailedException pulsarBatchListenerFailedException = (PulsarBatchListenerFailedException) exception;
Message<T> pulsarMessage = getPulsarMessageCausedTheException(pulsarBatchListenerFailedException);
final Message<T> theCurrentPulsarMessageTracked = this.pulsarConsumerErrorHandler.currentMessage();

4
spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarDeadLetterPublishingRecoverer.java
View File

@@ -68,8 +68,8 @@ public class PulsarDeadLetterPublishingRecoverer<T> implements PulsarMessageReco
this.pulsarTemplate.newMessage(message.getValue())
.withTopic(this.destinationResolver.apply(consumer, message))
.withMessageCustomizer(messageBuilder -> messageBuilder.property(EXCEPTION_THROWN_CAUSE,
exception.getCause() == null ? exception.getMessage()
: exception.getCause().getMessage()))
exception.getCause() != null ? exception.getCause().getMessage()
: exception.getMessage()))
.sendAsync();
}
catch (PulsarClientException e) {