From 2f7d4851fc3fb76579eea12dd3c387fb3d6bf548 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Sat, 10 Sep 2022 21:03:11 -0400 Subject: [PATCH] Introducing experimental consumer error handling (#100) * Introducing experimental consumer error handling Provide a Spring specific general mechanism to handle consumer errors regardless of the subscription type. Currently for exclusive and failover subscriptions, Pulsar Java client does not allow the applicaitons to use a DLQ. This feature allows the applications to set a PulsarConsumerErrorHandler that takes a Backoff and PulsarMessageRecoverer to retry and recover the failed message. This is an initial commit for this feature and more changes in this area will follow. Resolves https://github.com/spring-projects-experimental/spring-pulsar/issues/28 * Addressing PR review comments --- ...bstractPulsarMessageListenerContainer.java | 10 + .../DefaultPulsarConsumerErrorHandler.java | 102 ++++ ...DefaultPulsarMessageListenerContainer.java | 189 +++++- ...lsarBatchAcknowledgingMessageListener.java | 8 +- .../PulsarBatchListenerFailedException.java | 46 ++ .../listener/PulsarBatchMessageListener.java | 7 +- .../listener/PulsarConsumerErrorHandler.java | 63 ++ .../PulsarDeadLetterPublishingRecoverer.java | 80 +++ .../listener/PulsarMessageRecoverer.java | 46 ++ .../PulsarMessageRecovererFactory.java | 38 ++ ...rBatchMessagingMessageListenerAdapter.java | 22 +- .../PulsarBatchMessageConverter.java | 3 +- .../PulsarBatchMessagingMessageConverter.java | 3 +- .../core/ConsumerAcknowledgmentTests.java | 6 +- ...efaultPulsarConsumerErrorHandlerTests.java | 567 ++++++++++++++++++ 15 files changed, 1149 insertions(+), 41 deletions(-) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandler.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchListenerFailedException.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarConsumerErrorHandler.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarDeadLetterPublishingRecoverer.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageRecoverer.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageRecovererFactory.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java index 43388427..7fec5375 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java @@ -65,6 +65,8 @@ public abstract class AbstractPulsarMessageListenerContainer implements Pulsa protected DeadLetterPolicy deadLetterPolicy; + private PulsarConsumerErrorHandler pulsarConsumerErrorHandler; + @SuppressWarnings("unchecked") protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProperties) { @@ -198,4 +200,12 @@ public abstract class AbstractPulsarMessageListenerContainer implements Pulsa return this.deadLetterPolicy; } + public PulsarConsumerErrorHandler getPulsarConsumerErrorHandler() { + return this.pulsarConsumerErrorHandler; + } + + public void setPulsarConsumerErrorHandler(PulsarConsumerErrorHandler pulsarConsumerErrorHandler) { + this.pulsarConsumerErrorHandler = pulsarConsumerErrorHandler; + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandler.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandler.java new file mode 100644 index 00000000..f29ce419 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandler.java @@ -0,0 +1,102 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; + +import org.springframework.util.backoff.BackOff; +import org.springframework.util.backoff.BackOffExecution; + +/** + * Default implementation for {@link PulsarConsumerErrorHandler}. + *

+ * This implementation is capable for handling errors based on the interface contract. + * After handling the errors, if necessary, this implementation is capable of recovering + * the record(s) using a {@link PulsarMessageRecoverer} + * + * Note: This implementation uses a ThreadLocal to manage the current message in error and + * it's associated BackOffExecution. + * + * @param payload type managed by the Pulsar consumer + * @author Soby Chacko + */ +public class DefaultPulsarConsumerErrorHandler implements PulsarConsumerErrorHandler { + + private final PulsarMessageRecovererFactory pulsarMessageRecovererFactory; + + private final BackOff backOff; + + private final ThreadLocal backOffExecutionThreadLocal = new ThreadLocal<>(); + + public DefaultPulsarConsumerErrorHandler(PulsarMessageRecovererFactory pulsarMessageRecovererFactory, + BackOff backOff) { + this.pulsarMessageRecovererFactory = pulsarMessageRecovererFactory; + this.backOff = backOff; + } + + @Override + public boolean shouldRetryMessage(Exception exception, Message message) { + final Pair pair = this.backOffExecutionThreadLocal.get(); + long nextBackOff; + BackOffExecution backOffExecution; + if (pair != null && pair.message.equals(message)) { + backOffExecution = pair.backOffExecution; + } + else { + backOffExecution = this.backOff.start(); + this.backOffExecutionThreadLocal.set(new Pair(message, backOffExecution)); + } + nextBackOff = backOffExecution.nextBackOff(); + onNextBackoff(nextBackOff); + return nextBackOff != BackOffExecution.STOP; + } + + private void onNextBackoff(long nextBackOff) { + if (nextBackOff > BackOffExecution.STOP) { + try { + Thread.sleep(nextBackOff); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + + @Override + public void recoverMessage(Consumer consumer, Message message, Exception exception) { + this.pulsarMessageRecovererFactory.recovererForConsumer(consumer).recoverMessage(message, exception); + } + + @SuppressWarnings("unchecked") + public Message currentMessage() { + // there is only one message tracked at any time. + final Pair pair = this.backOffExecutionThreadLocal.get(); + if (pair == null) { + return null; + } + return (Message) pair.message(); + } + + public void clearMessage() { + this.backOffExecutionThreadLocal.remove(); + } + + private record Pair(Message message, BackOffExecution backOffExecution) { + }; + +} 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 037cfd63..1c41216d 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 @@ -16,6 +16,7 @@ package org.springframework.pulsar.listener; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; @@ -26,6 +27,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; import java.util.stream.Stream; import java.util.stream.StreamSupport; @@ -165,6 +167,8 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess private volatile Thread consumerThread; + private final PulsarConsumerErrorHandler pulsarConsumerErrorHandler; + @SuppressWarnings({ "unchecked", "rawtypes" }) Listener(MessageListener messageListener) { if (messageListener instanceof PulsarBatchMessageListener) { @@ -179,6 +183,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess this.listener = null; this.batchMessageListener = null; } + this.pulsarConsumerErrorHandler = getPulsarConsumerErrorHandler(); try { final PulsarContainerProperties pulsarContainerProperties = getPulsarContainerProperties(); Map propertiesToConsumer = extractDirectConsumerProperties(); @@ -257,27 +262,36 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess this.consumerThread = Thread.currentThread(); publishConsumerStartedEvent(); + AtomicBoolean inRetryMode = new AtomicBoolean(false); + AtomicBoolean messagesPendingInBatch = new AtomicBoolean(false); + Messages messages = null; + List> messageList = null; while (isRunning()) { - Messages messages = null; // Always receive messages in batch mode. try { - messages = this.consumer.batchReceive(); + if (!inRetryMode.get() && !messagesPendingInBatch.get()) { + messages = this.consumer.batchReceive(); + } } catch (PulsarClientException e) { DefaultPulsarMessageListenerContainer.this.logger.error(e, () -> "Error receiving messages."); } Assert.isTrue(messages != null, "Messages cannot be null."); if (this.containerProperties.isBatchListener()) { + if (!inRetryMode.get() && !messagesPendingInBatch.get()) { + messageList = new ArrayList<>(); + messages.forEach(messageList::add); + } try { - if (messages.size() > 0) { + if (messageList != null && messageList.size() > 0) { if (this.batchMessageListener instanceof PulsarBatchAcknowledgingMessageListener) { - this.batchMessageListener.received(this.consumer, messages, + this.batchMessageListener.received(this.consumer, messageList, this.containerProperties .getAckMode() == PulsarContainerProperties.AckMode.MANUAL ? new ConsumerBatchAcknowledgment(this.consumer) : null); } else { - this.batchMessageListener.received(this.consumer, messages); + this.batchMessageListener.received(this.consumer, messageList); } if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.BATCH) { try { @@ -295,38 +309,61 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess this.consumer.negativeAcknowledge(messages); } } + if (this.pulsarConsumerErrorHandler != null) { + pendingMessagesHandledSuccessfully(inRetryMode, messagesPendingInBatch); + } } } catch (Exception e) { - // the whole batch is negatively acknowledged in the event of an - // exception from the handler method. - this.consumer.negativeAcknowledge(messages); + if (this.pulsarConsumerErrorHandler != null) { + messageList = invokeBatchListenerErrorHandler(inRetryMode, messagesPendingInBatch, + messageList, e); + } + else { + // the whole batch is negatively acknowledged in the event of + // an exception from the handler method. + this.consumer.negativeAcknowledge(messages); + } } } else { for (Message message : messages) { - try { - if (this.listener instanceof PulsarAcknowledgingMessageListener) { - this.listener.received(this.consumer, message, - this.containerProperties - .getAckMode() == PulsarContainerProperties.AckMode.MANUAL - ? new ConsumerAcknowledgment(this.consumer, message) : null); + do { + try { + if (this.listener instanceof PulsarAcknowledgingMessageListener) { + this.listener.received(this.consumer, message, + this.containerProperties + .getAckMode() == PulsarContainerProperties.AckMode.MANUAL + ? new ConsumerAcknowledgment(this.consumer, message) + : null); + } + else if (this.listener != null) { + this.listener.received(this.consumer, message); + } + if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.RECORD) { + handleAck(message); + } + if (inRetryMode.get()) { + inRetryMode.set(false); + } } - else if (this.listener != null) { - this.listener.received(this.consumer, message); - } - if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.RECORD) { - handleAck(message); - } - } - catch (Exception e) { - if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.RECORD) { - this.consumer.negativeAcknowledge(message); - } - else if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.BATCH) { - this.nackableMessages.add(message.getMessageId()); + catch (Exception e) { + if (this.pulsarConsumerErrorHandler != null) { + invokeRecordListenerErrorHandler(inRetryMode, message, e); + } + else { + if (this.containerProperties + .getAckMode() == PulsarContainerProperties.AckMode.RECORD) { + this.consumer.negativeAcknowledge(message); + } + else if (this.containerProperties + .getAckMode() == PulsarContainerProperties.AckMode.BATCH) { + this.nackableMessages.add(message.getMessageId()); + } + } } } + while (inRetryMode.get()); } // All the records are processed at this point. Handle acks. if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.BATCH) { @@ -336,6 +373,104 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } } + /** + * Special scenario for batch error handling round1: messages m1,m2,...m10 are + * received batch listener throws error on m3 goes through error handle flow and + * tracks m3 and sets messgeList to m3,m4..m10 round2: in retry mode, no new + * messages received If at this point all messages are handled successfully then + * the normal flow will clear the handler state out. However, if the handler + * throws an error again it will be one of 2 things... m3 or a subsequent message + * m4-m10. + * @param inRetryMode is the message in retry mode + * @param messagesPendingInBatch are there pe nding messages from the batch + * @param messageList message list to process + * @param exception exception from the failed message + * @return a list of messages to be processed next. + */ + private List> invokeBatchListenerErrorHandler(AtomicBoolean inRetryMode, + AtomicBoolean messagesPendingInBatch, List> messageList, Exception exception) { + Assert.isInstanceOf(PulsarBatchListenerFailedException.class, exception, + "Batch listener should throw PulsarBatchListenerFailedException on errors."); + PulsarBatchListenerFailedException pulsarBatchListenerFailedException = (PulsarBatchListenerFailedException) exception; + Message pulsarMessage = getPulsarMessageCausedTheException(pulsarBatchListenerFailedException); + final Message theCurrentPulsarMessageTracked = this.pulsarConsumerErrorHandler.currentMessage(); + // Previous message in error handled during retry but another msg in sublist + // caused error; + // resetting state in order to track it + if (theCurrentPulsarMessageTracked != null && !theCurrentPulsarMessageTracked.equals(pulsarMessage)) { + pendingMessagesHandledSuccessfully(inRetryMode, messagesPendingInBatch); + } + // this is key to understanding how the message gets retried, it gets put into + // the new sublist + // at position 0 (aka it will be the 1st one re-sent to the listener and see + // if it can be + // handled on the retry. Otherwise, if we are out of retries then the sublist + // does not include + // the message in error (it instead gets recovered). + final int indexOfFailedMessage = messageList.indexOf(pulsarMessage); + messageList = messageList.subList(indexOfFailedMessage, messageList.size()); + final boolean toBeRetried = this.pulsarConsumerErrorHandler + .shouldRetryMessage(pulsarBatchListenerFailedException, pulsarMessage); + if (toBeRetried) { + inRetryMode.set(true); + } + else { + if (inRetryMode.get()) { + inRetryMode.set(false); + } + // retries exhausted - recover the message + this.pulsarConsumerErrorHandler.recoverMessage(this.consumer, pulsarMessage, + pulsarBatchListenerFailedException); + handleAck(pulsarMessage); + if (messageList.size() == 1) { + messagesPendingInBatch.set(false); + } + else { + messageList = messageList.subList(1, messageList.size()); + } + if (!messageList.isEmpty()) { + messagesPendingInBatch.set(true); + } + this.pulsarConsumerErrorHandler.clearMessage(); + } + return messageList; + } + + private void invokeRecordListenerErrorHandler(AtomicBoolean inRetryMode, Message message, Exception e) { + final boolean toBeRetried = this.pulsarConsumerErrorHandler.shouldRetryMessage(e, message); + if (toBeRetried) { + inRetryMode.set(true); + } + else { + if (inRetryMode.get()) { + inRetryMode.set(false); + } + // retries exhausted - recover the message + this.pulsarConsumerErrorHandler.recoverMessage(this.consumer, message, e); + // retries exhausted - if record ackmode, acknowledge, otherwise normal + // batch ack at the end + if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.RECORD) { + handleAck(message); + } + } + } + + private void pendingMessagesHandledSuccessfully(AtomicBoolean inRetryMode, + AtomicBoolean messagesPendingInBatch) { + if (inRetryMode.get()) { + inRetryMode.set(false); + } + if (messagesPendingInBatch.get()) { + messagesPendingInBatch.set(false); + } + this.pulsarConsumerErrorHandler.clearMessage(); + } + + @SuppressWarnings("unchecked") + private Message getPulsarMessageCausedTheException(PulsarBatchListenerFailedException exception) { + return (Message) exception.getMessageInError(); + } + private boolean isSharedSubscriptionType() { return this.containerProperties.getSubscriptionType() == SubscriptionType.Shared || this.containerProperties.getSubscriptionType() == SubscriptionType.Key_Shared; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchAcknowledgingMessageListener.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchAcknowledgingMessageListener.java index 856f6c05..4774f27b 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchAcknowledgingMessageListener.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchAcknowledgingMessageListener.java @@ -16,8 +16,10 @@ package org.springframework.pulsar.listener; +import java.util.List; + import org.apache.pulsar.client.api.Consumer; -import org.apache.pulsar.client.api.Messages; +import org.apache.pulsar.client.api.Message; /** * Batch message listener that allows manual acknowledgment. @@ -27,10 +29,10 @@ import org.apache.pulsar.client.api.Messages; */ public interface PulsarBatchAcknowledgingMessageListener extends PulsarBatchMessageListener { - default void received(Consumer consumer, Messages msg) { + default void received(Consumer consumer, List> msg) { throw new UnsupportedOperationException("Not Supported."); } - void received(Consumer consumer, Messages msg, Acknowledgement acknowledgement); + void received(Consumer consumer, List> msg, Acknowledgement acknowledgement); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchListenerFailedException.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchListenerFailedException.java new file mode 100644 index 00000000..1030a502 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchListenerFailedException.java @@ -0,0 +1,46 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +import org.apache.pulsar.client.api.Message; + +import org.springframework.pulsar.PulsarException; + +/** + * Batch message listeners should throw this exception in the event of an error. + * + * @author Soby Chacko + */ +public class PulsarBatchListenerFailedException extends PulsarException { + + private final Object messageInError; + + public PulsarBatchListenerFailedException(String msg, Message message) { + super(msg); + this.messageInError = message; + } + + public PulsarBatchListenerFailedException(String msg, Throwable cause, Message message) { + super(msg, cause); + this.messageInError = message; + } + + public Object getMessageInError() { + return this.messageInError; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchMessageListener.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchMessageListener.java index daea374d..4ef45332 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchMessageListener.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchMessageListener.java @@ -19,9 +19,10 @@ */ package org.springframework.pulsar.listener; +import java.util.List; + import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; -import org.apache.pulsar.client.api.Messages; /** * @@ -36,10 +37,10 @@ public interface PulsarBatchMessageListener extends PulsarRecordMessageListen throw new UnsupportedOperationException(); } - default void received(Consumer consumer, Messages msg, Acknowledgement acknowledgement) { + default void received(Consumer consumer, List> msg, Acknowledgement acknowledgement) { throw new UnsupportedOperationException(); } - void received(Consumer consumer, Messages msg); + void received(Consumer consumer, List> msg); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarConsumerErrorHandler.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarConsumerErrorHandler.java new file mode 100644 index 00000000..8bc2ffbe --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarConsumerErrorHandler.java @@ -0,0 +1,63 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; + +/** + * + * Contract for consumer error handling through the message listener container. Both + * record and batch message listener errors are handled through this interface. + * + * When an error handler implementation is provided to the message listener container, the + * container will funnel all errors through it for handling. + * + * @param payload type managed by the consumer + * @author Soby Chacko + */ +public interface PulsarConsumerErrorHandler { + + /** + * Decide if the failed message should be retried. + * @param exception throws exception + * @param message Pulsar message + * @return if the failed message should be retried or not + */ + boolean shouldRetryMessage(Exception exception, Message message); + + /** + * Recover the message based on the implementation provided. Once this method returns, + * callers can assume that the message is recovered and has not been acknowledged yet. + * @param consumer Pulsar consumer + * @param message Pulsar message + * @param thrownException thrown exception + */ + void recoverMessage(Consumer consumer, Message message, Exception thrownException); + + /** + * Returns the current message in error. + * @return the Pulsar Message currently tracked by the error handler + */ + Message currentMessage(); + + /** + * Clear the message in error from managing (such as resetting any thread state etc.). + */ + void clearMessage(); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarDeadLetterPublishingRecoverer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarDeadLetterPublishingRecoverer.java new file mode 100644 index 00000000..628a79a1 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarDeadLetterPublishingRecoverer.java @@ -0,0 +1,80 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +import java.util.function.BiFunction; + +import org.apache.commons.logging.LogFactory; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.PulsarClientException; + +import org.springframework.core.log.LogAccessor; +import org.springframework.pulsar.core.PulsarOperations; + +/** + * {@link PulsarMessageRecoverer} implementation that is capable of recovering the message + * by publishing the failed record to a DLT - Dead Letter Topic. + * + * @param payload type of the Pulsar message + * @author Soby Chacko + */ +public class PulsarDeadLetterPublishingRecoverer implements PulsarMessageRecovererFactory { + + protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); + + /** + * TODO: Move this to a common constants class. + * + * exception cause for the failed message. + */ + public static final String EXCEPTION_THROWN_CAUSE = "exception-thrown-cause"; + + private static final BiFunction, Message, String> DEFAULT_DESTINATION_RESOLVER = (c, + m) -> m.getTopicName() + "-" + c.getSubscription() + "-DLT"; + + private final PulsarOperations pulsarTemplate; + + private final BiFunction, Message, String> destinationResolver; + + public PulsarDeadLetterPublishingRecoverer(PulsarOperations pulsarTemplate) { + this(pulsarTemplate, DEFAULT_DESTINATION_RESOLVER); + } + + public PulsarDeadLetterPublishingRecoverer(PulsarOperations pulsarTemplate, + BiFunction, Message, String> destinationResolver) { + this.pulsarTemplate = pulsarTemplate; + this.destinationResolver = destinationResolver; + } + + @Override + public PulsarMessageRecoverer recovererForConsumer(Consumer consumer) { + return (message, exception) -> { + try { + this.pulsarTemplate.newMessage(message.getValue()) + .withTopic(this.destinationResolver.apply(consumer, message)) + .withMessageCustomizer(messageBuilder -> messageBuilder.property(EXCEPTION_THROWN_CAUSE, + exception.getCause().getMessage())) + .sendAsync(); + } + catch (PulsarClientException e) { + this.logger.error(e, "DLT publishing failed."); + } + }; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageRecoverer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageRecoverer.java new file mode 100644 index 00000000..d7db6193 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageRecoverer.java @@ -0,0 +1,46 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +import java.util.function.BiConsumer; + +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; + +/** + * Allows recovering a failed Pulsar message. + * + * Implementations can choose how the message needs to be recovered by providing a + * {@link java.util.function.Function} implementation that takes a {@link Consumer} and + * then provide a {@link BiConsumer} which takes {@link Message} and the thrown + * {@link Exception}. + * + * @param payload type of Pulsar message. + * @author Soby Chacko + * @author Chris Bono + */ +@FunctionalInterface +public interface PulsarMessageRecoverer { + + /** + * Recover a failed message, for e.g. send the message to a DLT. + * @param message Pulsar message + * @param exception exception from failed message + */ + void recoverMessage(Message message, Exception exception); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageRecovererFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageRecovererFactory.java new file mode 100644 index 00000000..9a4be53c --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageRecovererFactory.java @@ -0,0 +1,38 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +import org.apache.pulsar.client.api.Consumer; + +/** + * Factory interface for {@link PulsarMessageRecoverer}. + * + * @param message type + * @author Soby Chacko + * @author Chris Bono + */ +@FunctionalInterface +public interface PulsarMessageRecovererFactory { + + /** + * Provides a message recoverer {@link PulsarMessageRecoverer}. + * @param consumer Pulsar consumer + * @return {@link PulsarMessageRecoverer}. + */ + PulsarMessageRecoverer recovererForConsumer(Consumer consumer); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java index 65372b4d..82f2a278 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java @@ -18,6 +18,7 @@ package org.springframework.pulsar.listener.adapter; import java.lang.reflect.Method; import java.util.ArrayList; +import java.util.Iterator; import java.util.List; import org.apache.pulsar.client.api.Consumer; @@ -65,7 +66,8 @@ public class PulsarBatchMessagingMessageListenerAdapter extends PulsarMessagi } @Override - public void received(Consumer consumer, Messages msg, @Nullable Acknowledgement acknowledgement) { + public void received(Consumer consumer, List> msg, + @Nullable Acknowledgement acknowledgement) { Message message; if (!isConsumerRecordList()) { if (isMessageList()) { @@ -83,7 +85,21 @@ public class PulsarBatchMessagingMessageListenerAdapter extends PulsarMessagi message = null; // optimization since we won't need any conversion to invoke } logger.debug(() -> "Processing [" + message + "]"); - invoke(msg, consumer, message, acknowledgement); + + // In order to avoid clash with target List payload type. + final Messages messages = new Messages<>() { + + @Override + public Iterator> iterator() { + return msg.iterator(); + } + + @Override + public int size() { + return msg.size(); + } + }; + invoke(messages, consumer, message, acknowledgement); } protected void invoke(Object records, Consumer consumer, final Message messageArg, @@ -101,7 +117,7 @@ public class PulsarBatchMessagingMessageListenerAdapter extends PulsarMessagi } } - protected Message toMessagingMessage(Messages msg, Consumer consumer) { + protected Message toMessagingMessage(List> msg, Consumer consumer) { return getBatchMessageConverter().toMessage(msg, consumer, getType()); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessageConverter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessageConverter.java index a9f3ff53..e7515104 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessageConverter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessageConverter.java @@ -17,6 +17,7 @@ package org.springframework.pulsar.support.converter; import java.lang.reflect.Type; +import java.util.List; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Messages; @@ -32,7 +33,7 @@ import org.springframework.pulsar.support.MessageConverter; */ public interface PulsarBatchMessageConverter extends MessageConverter { - Message toMessage(Messages records, Consumer consumer, Type payloadType); + Message toMessage(List> msg, Consumer consumer, Type payloadType); T fromMessage(Messages message, String defaultTopic); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessagingMessageConverter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessagingMessageConverter.java index 8c784250..5a42e7c3 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessagingMessageConverter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessagingMessageConverter.java @@ -48,7 +48,8 @@ public class PulsarBatchMessagingMessageConverter implements PulsarBatchMessa } @Override - public Message toMessage(Messages records, Consumer consumer, Type type) { + public Message toMessage(List> records, Consumer consumer, + Type type) { List payloads = new ArrayList<>(); List conversionFailures = new ArrayList<>(); for (org.apache.pulsar.client.api.Message message : records) { diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java index e6c80291..8df8b591 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java @@ -289,7 +289,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { doAnswer(invocation -> { latch.countDown(); return null; - }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(Messages.class)); + }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class)); pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.STRING); @@ -308,7 +308,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { } assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue(); await().atMost(Duration.ofSeconds(10)).untilAsserted( - () -> verify(pulsarBatchMessageListener, times(1)).received(any(Consumer.class), any(Messages.class))); + () -> verify(pulsarBatchMessageListener, times(1)).received(any(Consumer.class), any(List.class))); await().atMost(Duration.ofSeconds(10)) .untilAsserted(() -> verify(containerConsumer, times(1)).acknowledgeCumulative(any(Message.class))); container.stop(); @@ -337,7 +337,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { doAnswer(invocation -> { latch.countDown(); throw new RuntimeException(); - }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(Messages.class)); + }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class)); pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.STRING); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java new file mode 100644 index 00000000..3e9a4f82 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java @@ -0,0 +1,567 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +import static org.awaitility.Awaitility.await; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Schema; +import org.junit.jupiter.api.Test; + +import org.springframework.pulsar.core.AbstractContainerBaseTests; +import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; +import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.PulsarOperations; +import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.TypedMessageBuilderCustomizer; +import org.springframework.util.backoff.FixedBackOff; + +/** + * @author Soby Chacko + */ +public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBaseTests { + + @Test + @SuppressWarnings("unchecked") + void happyPathErrorHandlingForRecordMessageListener() throws Exception { + Map config = new HashMap<>(); + config.put("topicNames", Collections.singleton("default-error-handler-tests-1")); + config.put("subscriptionName", "default-error-handler-tests-sub-1"); + + final PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build(); + final DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( + pulsarClient, config); + + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + PulsarRecordMessageListener messageListener = mock(PulsarRecordMessageListener.class); + + doAnswer(invocation -> { + throw new RuntimeException(); + }).when(messageListener).received(any(Consumer.class), any(Message.class)); + + pulsarContainerProperties.setMessageListener(messageListener); + pulsarContainerProperties.setSchema(Schema.STRING); + + Map prodConfig = new HashMap<>(); + prodConfig.put("topicName", "default-error-handler-tests-1"); + DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + prodConfig); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); + + DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>( + new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10))); + container.start(); + + pulsarTemplate.sendAsync("hello john doe"); + + PulsarOperations.SendMessageBuilder sendMessageBuilderMock = mock( + PulsarOperations.SendMessageBuilder.class); + + when(mockPulsarTemplate.newMessage("hello john doe").withTopic(any(String.class)) + .withMessageCustomizer(any(TypedMessageBuilderCustomizer.class))).thenReturn(sendMessageBuilderMock); + + await().atMost(Duration.ofSeconds(10)).untilAsserted( + () -> verify(messageListener, times(11)).received(any(Consumer.class), any(Message.class))); + await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> verify(sendMessageBuilderMock).sendAsync()); + + container.stop(); + pulsarClient.close(); + } + + @Test + @SuppressWarnings("unchecked") + void errorHandlingForRecordMessageListenerWithTransientError() throws Exception { + Map config = new HashMap<>(); + config.put("topicNames", Collections.singleton("default-error-handler-tests-2")); + config.put("subscriptionName", "default-error-handler-tests-sub-2"); + + final PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build(); + final DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( + pulsarClient, config); + + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + PulsarRecordMessageListener messageListener = mock(PulsarRecordMessageListener.class); + AtomicInteger count = new AtomicInteger(0); + doAnswer(invocation -> { + final int currentCount = count.incrementAndGet(); + if (currentCount <= 3) { + throw new RuntimeException(); + } + return new Object(); + }).when(messageListener).received(any(Consumer.class), any(Message.class)); + + pulsarContainerProperties.setMessageListener(messageListener); + pulsarContainerProperties.setSchema(Schema.STRING); + + Map prodConfig = new HashMap<>(); + prodConfig.put("topicName", "default-error-handler-tests-2"); + DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + prodConfig); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class); + + DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>( + new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10))); + container.start(); + + pulsarTemplate.sendAsync("hello john doe"); + + await().atMost(Duration.ofSeconds(10)).untilAsserted( + () -> verify(messageListener, times(4)).received(any(Consumer.class), any(Message.class))); + verifyNoInteractions(mockPulsarTemplate); + + container.stop(); + pulsarClient.close(); + } + + @Test + @SuppressWarnings("unchecked") + void everyOtherRecordThrowsNonTransientExceptionsRecordMessageListener() throws Exception { + Map config = new HashMap<>(); + config.put("topicNames", Collections.singleton("default-error-handler-tests-3")); + config.put("subscriptionName", "default-error-handler-tests-sub-3"); + + final PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build(); + final DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( + pulsarClient, config); + + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + PulsarRecordMessageListener messageListener = mock(PulsarRecordMessageListener.class); + doAnswer(invocation -> { + final Message message = invocation.getArgument(1); + final Integer value = message.getValue(); + if (value % 2 == 0) { + throw new RuntimeException(); + } + return new Object(); + }).when(messageListener).received(any(Consumer.class), any(Message.class)); + + pulsarContainerProperties.setMessageListener(messageListener); + pulsarContainerProperties.setSchema(Schema.INT32); + + Map prodConfig = new HashMap<>(); + prodConfig.put("topicName", "default-error-handler-tests-3"); + DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + prodConfig); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); + + DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>( + new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 5))); + container.start(); + + for (int i = 0; i < 10; i++) { + pulsarTemplate.sendAsync(i); + } + + PulsarOperations.SendMessageBuilder sendMessageBuilderMock = mock( + PulsarOperations.SendMessageBuilder.class); + + when(mockPulsarTemplate.newMessage(any(Integer.class)).withTopic(any(String.class)) + .withMessageCustomizer(any(TypedMessageBuilderCustomizer.class))).thenReturn(sendMessageBuilderMock); + + // 5 records fail - 5 * (1 + 5 max retry) = 30 + 5 records don't fail = 35 + await().atMost(Duration.ofSeconds(30)).untilAsserted( + () -> verify(messageListener, times(35)).received(any(Consumer.class), any(Message.class))); + await().atMost(Duration.ofSeconds(30)) + .untilAsserted(() -> verify(sendMessageBuilderMock, times(5)).sendAsync()); + + container.stop(); + pulsarClient.close(); + } + + @Test + @SuppressWarnings("unchecked") + void batchRecordListenerFirstOneOnlyErrorAndRecover() throws Exception { + Map config = new HashMap<>(); + config.put("topicNames", Collections.singleton("default-error-handler-tests-4")); + config.put("subscriptionName", "default-error-handler-tests-sub-4"); + + final PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build(); + final DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( + pulsarClient, config); + + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + pulsarContainerProperties.setMaxNumMessages(10); + pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchListener(true); + final PulsarBatchAcknowledgingMessageListener pulsarBatchMessageListener = mock( + PulsarBatchAcknowledgingMessageListener.class); + + doAnswer(invocation -> { + final List> message = invocation.getArgument(1); + final Message integerMessage = message.get(0); + final Integer value = integerMessage.getValue(); + if (value == 0) { + throw new PulsarBatchListenerFailedException("failed", integerMessage); + } + final Acknowledgement acknowledgment = invocation.getArgument(2); + List messageIds = new ArrayList<>(); + for (Message integerMessage1 : message) { + messageIds.add(integerMessage1.getMessageId()); + } + acknowledgment.acknowledge(messageIds); + return new Object(); + }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class), any(Acknowledgement.class)); + + pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); + pulsarContainerProperties.setSchema(Schema.INT32); + pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); + + container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>( + new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10))); + + container.start(); + + Map prodConfig = new HashMap<>(); + prodConfig.put("topicName", "default-error-handler-tests-4"); + final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, prodConfig); + final PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + for (int i = 0; i < 10; i++) { + pulsarTemplate.sendAsync(i); + } + + PulsarOperations.SendMessageBuilder sendMessageBuilderMock = mock( + PulsarOperations.SendMessageBuilder.class); + + when(mockPulsarTemplate.newMessage(any(Integer.class)).withTopic(any(String.class)) + .withMessageCustomizer(any(TypedMessageBuilderCustomizer.class))).thenReturn(sendMessageBuilderMock); + + // 1 + 10 + 1 = 12 calls altogether + await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> verify(pulsarBatchMessageListener, times(12)) + .received(any(Consumer.class), any(List.class), any(Acknowledgement.class))); + await().atMost(Duration.ofSeconds(30)) + .untilAsserted(() -> verify(sendMessageBuilderMock, times(1)).sendAsync()); + + container.stop(); + pulsarClient.close(); + } + + @Test + @SuppressWarnings("unchecked") + void batchRecordListenerRecordFailsInTheMiddle() throws Exception { + Map config = new HashMap<>(); + config.put("topicNames", Collections.singleton("default-error-handler-tests-5")); + config.put("subscriptionName", "default-error-handler-tests-sub-5"); + + final PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build(); + final DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( + pulsarClient, config); + + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + pulsarContainerProperties.setMaxNumMessages(10); + pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchListener(true); + final PulsarBatchAcknowledgingMessageListener pulsarBatchMessageListener = mock( + PulsarBatchAcknowledgingMessageListener.class); + + doAnswer(invocation -> { + final List> messages = invocation.getArgument(1); + + for (Message message : messages) { + if (message.getValue() == 5) { + throw new PulsarBatchListenerFailedException("failed", message); + } + else { + final Acknowledgement acknowledgment = invocation.getArgument(2); + acknowledgment.acknowledge(message.getMessageId()); + } + } + return new Object(); + }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class), any(Acknowledgement.class)); + + pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); + pulsarContainerProperties.setSchema(Schema.INT32); + pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); + + container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>( + new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10))); + + container.start(); + + Map prodConfig = new HashMap<>(); + prodConfig.put("topicName", "default-error-handler-tests-5"); + final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, prodConfig); + final PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + for (int i = 0; i < 10; i++) { + pulsarTemplate.sendAsync(i); + } + PulsarOperations.SendMessageBuilder sendMessageBuilderMock = mock( + PulsarOperations.SendMessageBuilder.class); + + when(mockPulsarTemplate.newMessage(any(Integer.class)).withTopic(any(String.class)) + .withMessageCustomizer(any(TypedMessageBuilderCustomizer.class))).thenReturn(sendMessageBuilderMock); + + // 1 + 10 + 1 = 12 calls altogether + await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> verify(pulsarBatchMessageListener, times(12)) + .received(any(Consumer.class), any(List.class), any(Acknowledgement.class))); + await().atMost(Duration.ofSeconds(30)) + .untilAsserted(() -> verify(sendMessageBuilderMock, times(1)).sendAsync()); + + container.stop(); + pulsarClient.close(); + } + + @Test + @SuppressWarnings("unchecked") + void batchRecordListenerRecordFailsTwiceInTheMiddle() throws Exception { + Map config = new HashMap<>(); + config.put("topicNames", Collections.singleton("default-error-handler-tests-6")); + config.put("subscriptionName", "default-error-handler-tests-sub-6"); + + final PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build(); + final DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( + pulsarClient, config); + + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + pulsarContainerProperties.setMaxNumMessages(10); + pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchListener(true); + final PulsarBatchAcknowledgingMessageListener pulsarBatchMessageListener = mock( + PulsarBatchAcknowledgingMessageListener.class); + + doAnswer(invocation -> { + final List> messages = invocation.getArgument(1); + + for (Message message : messages) { + if (message.getValue() == 2 || message.getValue() == 5) { + throw new PulsarBatchListenerFailedException("failed", message); + } + else { + final Acknowledgement acknowledgment = invocation.getArgument(2); + acknowledgment.acknowledge(message.getMessageId()); + } + } + return new Object(); + }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class), any(Acknowledgement.class)); + + pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); + pulsarContainerProperties.setSchema(Schema.INT32); + pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); + + container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>( + new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10))); + + container.start(); + + Map prodConfig = new HashMap<>(); + prodConfig.put("topicName", "default-error-handler-tests-6"); + final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, prodConfig); + final PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + for (int i = 0; i < 10; i++) { + pulsarTemplate.sendAsync(i); + } + PulsarOperations.SendMessageBuilder sendMessageBuilderMock = mock( + PulsarOperations.SendMessageBuilder.class); + + when(mockPulsarTemplate.newMessage(any(Integer.class)).withTopic(any(String.class)) + .withMessageCustomizer(any(TypedMessageBuilderCustomizer.class))).thenReturn(sendMessageBuilderMock); + + // 1 + 10 + 1 + 10 + 1 = 23 calls altogether + await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> verify(pulsarBatchMessageListener, times(23)) + .received(any(Consumer.class), any(List.class), any(Acknowledgement.class))); + await().atMost(Duration.ofSeconds(30)) + .untilAsserted(() -> verify(sendMessageBuilderMock, times(2)).sendAsync()); + + container.stop(); + pulsarClient.close(); + } + + @Test + @SuppressWarnings("unchecked") + void batchRecordListenerRecordFailsInTheMiddleButTransientError() throws Exception { + Map config = new HashMap<>(); + config.put("topicNames", Collections.singleton("default-error-handler-tests-7")); + config.put("subscriptionName", "default-error-handler-tests-sub-7"); + + final PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build(); + final DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( + pulsarClient, config); + + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + pulsarContainerProperties.setMaxNumMessages(10); + pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchListener(true); + final PulsarBatchAcknowledgingMessageListener pulsarBatchMessageListener = mock( + PulsarBatchAcknowledgingMessageListener.class); + + AtomicInteger count = new AtomicInteger(0); + doAnswer(invocation -> { + final List> messages = invocation.getArgument(1); + final Acknowledgement acknowledgment = invocation.getArgument(2); + for (Message message : messages) { + if (message.getValue() == 5) { + final int currentCount = count.getAndIncrement(); + if (currentCount < 3) { + throw new PulsarBatchListenerFailedException("failed", message); + } + else { + acknowledgment.acknowledge(message.getMessageId()); + } + } + else { + acknowledgment.acknowledge(message.getMessageId()); + } + } + return new Object(); + }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class), any(Acknowledgement.class)); + + pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); + pulsarContainerProperties.setSchema(Schema.INT32); + pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); + + container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>( + new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10))); + + container.start(); + + Map prodConfig = new HashMap<>(); + prodConfig.put("topicName", "default-error-handler-tests-7"); + final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, prodConfig); + final PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + for (int i = 0; i < 10; i++) { + pulsarTemplate.sendAsync(i); + } + // 1 + 3 + 1 = 5 calls altogether + await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> verify(pulsarBatchMessageListener, times(4)) + .received(any(Consumer.class), any(List.class), any(Acknowledgement.class))); + verifyNoInteractions(mockPulsarTemplate); + + container.stop(); + pulsarClient.close(); + } + + @Test + @SuppressWarnings("unchecked") + void batchListenerFailsTransientErrorFollowedByNonTransient() throws Exception { + Map config = new HashMap<>(); + config.put("topicNames", Collections.singleton("default-error-handler-tests-8")); + config.put("subscriptionName", "default-error-handler-tests-sub-8"); + + final PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build(); + final DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( + pulsarClient, config); + + PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + pulsarContainerProperties.setMaxNumMessages(10); + pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchListener(true); + final PulsarBatchAcknowledgingMessageListener pulsarBatchMessageListener = mock( + PulsarBatchAcknowledgingMessageListener.class); + + AtomicInteger count = new AtomicInteger(0); + doAnswer(invocation -> { + final List> messages = invocation.getArgument(1); + final Acknowledgement acknowledgment = invocation.getArgument(2); + for (Message message : messages) { + if (message.getValue() == 5) { + final int currentCount = count.getAndIncrement(); + if (currentCount < 3) { + throw new PulsarBatchListenerFailedException("failed", message); + } + else { + acknowledgment.acknowledge(message.getMessageId()); + } + } + else if (message.getValue() == 7) { + throw new PulsarBatchListenerFailedException("failed", message); + } + else { + acknowledgment.acknowledge(message.getMessageId()); + } + } + return new Object(); + }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class), any(Acknowledgement.class)); + + pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); + pulsarContainerProperties.setSchema(Schema.INT32); + pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL); + DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( + pulsarConsumerFactory, pulsarContainerProperties); + PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); + + container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>( + new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10))); + + container.start(); + + Map prodConfig = new HashMap<>(); + prodConfig.put("topicName", "default-error-handler-tests-8"); + final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, prodConfig); + final PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + for (int i = 0; i < 10; i++) { + pulsarTemplate.sendAsync(i); + } + PulsarOperations.SendMessageBuilder sendMessageBuilderMock = mock( + PulsarOperations.SendMessageBuilder.class); + + when(mockPulsarTemplate.newMessage(any(Integer.class)).withTopic(any(String.class)) + .withMessageCustomizer(any(TypedMessageBuilderCustomizer.class))).thenReturn(sendMessageBuilderMock); + // 1 + 2 + 1 + 10 + 1 = 15 calls altogether + await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> verify(pulsarBatchMessageListener, times(15)) + .received(any(Consumer.class), any(List.class), any(Acknowledgement.class))); + await().atMost(Duration.ofSeconds(30)) + .untilAsserted(() -> verify(sendMessageBuilderMock, times(1)).sendAsync()); + + container.stop(); + pulsarClient.close(); + } + +}