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

spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java

@@ -65,6 +65,8 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
 
 	protected DeadLetterPolicy deadLetterPolicy;
 
+	private PulsarConsumerErrorHandler<T> pulsarConsumerErrorHandler;
+
 	@SuppressWarnings("unchecked")
 	protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
 			PulsarContainerProperties pulsarContainerProperties) {
@@ -198,4 +200,12 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
 		return this.deadLetterPolicy;
 	}
 
+	public PulsarConsumerErrorHandler<T> getPulsarConsumerErrorHandler() {
+		return this.pulsarConsumerErrorHandler;
+	}
+
+	public void setPulsarConsumerErrorHandler(PulsarConsumerErrorHandler<T> pulsarConsumerErrorHandler) {
+		this.pulsarConsumerErrorHandler = pulsarConsumerErrorHandler;
+	}
+
 }

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}.
+ * <p>
+ * 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 <T> payload type managed by the Pulsar consumer
+ * @author Soby Chacko
+ */
+public class DefaultPulsarConsumerErrorHandler<T> implements PulsarConsumerErrorHandler<T> {
+
+	private final PulsarMessageRecovererFactory<T> pulsarMessageRecovererFactory;
+
+	private final BackOff backOff;
+
+	private final ThreadLocal<Pair> backOffExecutionThreadLocal = new ThreadLocal<>();
+
+	public DefaultPulsarConsumerErrorHandler(PulsarMessageRecovererFactory<T> pulsarMessageRecovererFactory,
+			BackOff backOff) {
+		this.pulsarMessageRecovererFactory = pulsarMessageRecovererFactory;
+		this.backOff = backOff;
+	}
+
+	@Override
+	public boolean shouldRetryMessage(Exception exception, Message<T> 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<T> consumer, Message<T> message, Exception exception) {
+		this.pulsarMessageRecovererFactory.recovererForConsumer(consumer).recoverMessage(message, exception);
+	}
+
+	@SuppressWarnings("unchecked")
+	public Message<T> currentMessage() {
+		// there is only one message tracked at any time.
+		final Pair pair = this.backOffExecutionThreadLocal.get();
+		if (pair == null) {
+			return null;
+		}
+		return (Message<T>) pair.message();
+	}
+
+	public void clearMessage() {
+		this.backOffExecutionThreadLocal.remove();
+	}
+
+	private record Pair(Message<?> message, BackOffExecution backOffExecution) {
+	};
+
+}

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<T> extends AbstractPulsarMess
 
 		private volatile Thread consumerThread;
 
+		private final PulsarConsumerErrorHandler<T> pulsarConsumerErrorHandler;
+
 		@SuppressWarnings({ "unchecked", "rawtypes" })
 		Listener(MessageListener<?> messageListener) {
 			if (messageListener instanceof PulsarBatchMessageListener) {
@@ -179,6 +183,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 				this.listener = null;
 				this.batchMessageListener = null;
 			}
+			this.pulsarConsumerErrorHandler = getPulsarConsumerErrorHandler();
 			try {
 				final PulsarContainerProperties pulsarContainerProperties = getPulsarContainerProperties();
 				Map<String, Object> propertiesToConsumer = extractDirectConsumerProperties();
@@ -257,27 +262,36 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 			this.consumerThread = Thread.currentThread();
 
 			publishConsumerStartedEvent();
+			AtomicBoolean inRetryMode = new AtomicBoolean(false);
+			AtomicBoolean messagesPendingInBatch = new AtomicBoolean(false);
+			Messages<T> messages = null;
+			List<Message<T>> messageList = null;
 			while (isRunning()) {
-				Messages<T> 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<T> 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<T> 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<T> 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<Message<T>> invokeBatchListenerErrorHandler(AtomicBoolean inRetryMode,
+				AtomicBoolean messagesPendingInBatch, List<Message<T>> messageList, Exception exception) {
+			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();
+			// 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<T> 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<T> getPulsarMessageCausedTheException(PulsarBatchListenerFailedException exception) {
+			return (Message<T>) exception.getMessageInError();
+		}
+
 		private boolean isSharedSubscriptionType() {
 			return this.containerProperties.getSubscriptionType() == SubscriptionType.Shared
 					|| this.containerProperties.getSubscriptionType() == SubscriptionType.Key_Shared;

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<T> extends PulsarBatchMessageListener<T> {
 
-	default void received(Consumer<T> consumer, Messages<T> msg) {
+	default void received(Consumer<T> consumer, List<Message<T>> msg) {
 		throw new UnsupportedOperationException("Not Supported.");
 	}
 
-	void received(Consumer<T> consumer, Messages<T> msg, Acknowledgement acknowledgement);
+	void received(Consumer<T> consumer, List<Message<T>> msg, Acknowledgement acknowledgement);
 
 }

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;
+	}
+
+}

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<T> extends PulsarRecordMessageListen
 		throw new UnsupportedOperationException();
 	}
 
-	default void received(Consumer<T> consumer, Messages<T> msg, Acknowledgement acknowledgement) {
+	default void received(Consumer<T> consumer, List<Message<T>> msg, Acknowledgement acknowledgement) {
 		throw new UnsupportedOperationException();
 	}
 
-	void received(Consumer<T> consumer, Messages<T> msg);
+	void received(Consumer<T> consumer, List<Message<T>> msg);
 
 }

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 <T> payload type managed by the consumer
+ * @author Soby Chacko
+ */
+public interface PulsarConsumerErrorHandler<T> {
+
+	/**
+	 * 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<T> 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<T> consumer, Message<T> message, Exception thrownException);
+
+	/**
+	 * Returns the current message in error.
+	 * @return the Pulsar Message currently tracked by the error handler
+	 */
+	Message<T> currentMessage();
+
+	/**
+	 * Clear the message in error from managing (such as resetting any thread state etc.).
+	 */
+	void clearMessage();
+
+}

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 <T> payload type of the Pulsar message
+ * @author Soby Chacko
+ */
+public class PulsarDeadLetterPublishingRecoverer<T> implements PulsarMessageRecovererFactory<T> {
+
+	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<Consumer<?>, Message<?>, String> DEFAULT_DESTINATION_RESOLVER = (c,
+			m) -> m.getTopicName() + "-" + c.getSubscription() + "-DLT";
+
+	private final PulsarOperations<T> pulsarTemplate;
+
+	private final BiFunction<Consumer<?>, Message<?>, String> destinationResolver;
+
+	public PulsarDeadLetterPublishingRecoverer(PulsarOperations<T> pulsarTemplate) {
+		this(pulsarTemplate, DEFAULT_DESTINATION_RESOLVER);
+	}
+
+	public PulsarDeadLetterPublishingRecoverer(PulsarOperations<T> pulsarTemplate,
+			BiFunction<Consumer<?>, Message<?>, String> destinationResolver) {
+		this.pulsarTemplate = pulsarTemplate;
+		this.destinationResolver = destinationResolver;
+	}
+
+	@Override
+	public PulsarMessageRecoverer<T> recovererForConsumer(Consumer<T> 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.");
+			}
+		};
+	}
+
+}

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 <T> payload type of Pulsar message.
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface PulsarMessageRecoverer<T> {
+
+	/**
+	 * 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<T> message, Exception exception);
+
+}

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 <T> message type
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface PulsarMessageRecovererFactory<T> {
+
+	/**
+	 * Provides a message recoverer {@link PulsarMessageRecoverer}.
+	 * @param consumer Pulsar consumer
+	 * @return {@link PulsarMessageRecoverer}.
+	 */
+	PulsarMessageRecoverer<T> recovererForConsumer(Consumer<T> consumer);
+
+}

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<V> extends PulsarMessagi
 	}
 
 	@Override
-	public void received(Consumer<V> consumer, Messages<V> msg, @Nullable Acknowledgement acknowledgement) {
+	public void received(Consumer<V> consumer, List<org.apache.pulsar.client.api.Message<V>> msg,
+			@Nullable Acknowledgement acknowledgement) {
 		Message<?> message;
 		if (!isConsumerRecordList()) {
 			if (isMessageList()) {
@@ -83,7 +85,21 @@ public class PulsarBatchMessagingMessageListenerAdapter<V> 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<V> messages = new Messages<>() {
+
+			@Override
+			public Iterator<org.apache.pulsar.client.api.Message<V>> iterator() {
+				return msg.iterator();
+			}
+
+			@Override
+			public int size() {
+				return msg.size();
+			}
+		};
+		invoke(messages, consumer, message, acknowledgement);
 	}
 
 	protected void invoke(Object records, Consumer<V> consumer, final Message<?> messageArg,
@@ -101,7 +117,7 @@ public class PulsarBatchMessagingMessageListenerAdapter<V> extends PulsarMessagi
 		}
 	}
 
-	protected Message<?> toMessagingMessage(Messages<V> msg, Consumer<V> consumer) {
+	protected Message<?> toMessagingMessage(List<org.apache.pulsar.client.api.Message<V>> msg, Consumer<V> consumer) {
 
 		return getBatchMessageConverter().toMessage(msg, consumer, getType());
 	}

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<T> extends MessageConverter {
 
-	Message<?> toMessage(Messages<T> records, Consumer<T> consumer, Type payloadType);
+	Message<?> toMessage(List<org.apache.pulsar.client.api.Message<T>> msg, Consumer<T> consumer, Type payloadType);
 
 	T fromMessage(Messages<T> message, String defaultTopic);
 

spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarBatchMessagingMessageConverter.java

@@ -48,7 +48,8 @@ public class PulsarBatchMessagingMessageConverter<T> implements PulsarBatchMessa
 	}
 
 	@Override
-	public Message<?> toMessage(Messages<T> records, Consumer<T> consumer, Type type) {
+	public Message<?> toMessage(List<org.apache.pulsar.client.api.Message<T>> records, Consumer<T> consumer,
+			Type type) {
 		List<Object> payloads = new ArrayList<>();
 		List<Exception> conversionFailures = new ArrayList<>();
 		for (org.apache.pulsar.client.api.Message<T> message : records) {

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);

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<String, Object> 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<String> 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<String, Object> prodConfig = new HashMap<>();
+		prodConfig.put("topicName", "default-error-handler-tests-1");
+		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+				prodConfig);
+		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+		PulsarTemplate<String> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
+
+		DefaultPulsarMessageListenerContainer<String> 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<String> 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<String, Object> 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<String> 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<String, Object> prodConfig = new HashMap<>();
+		prodConfig.put("topicName", "default-error-handler-tests-2");
+		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+				prodConfig);
+		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+		PulsarTemplate<String> mockPulsarTemplate = mock(PulsarTemplate.class);
+
+		DefaultPulsarMessageListenerContainer<String> 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<String, Object> 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<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
+				pulsarClient, config);
+
+		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
+		PulsarRecordMessageListener<?> messageListener = mock(PulsarRecordMessageListener.class);
+		doAnswer(invocation -> {
+			final Message<Integer> 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<String, Object> prodConfig = new HashMap<>();
+		prodConfig.put("topicName", "default-error-handler-tests-3");
+		DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+				prodConfig);
+		PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+		PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
+
+		DefaultPulsarMessageListenerContainer<Integer> 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<Integer> 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<String, Object> 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<Integer> 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<Integer>> message = invocation.getArgument(1);
+			final Message<Integer> integerMessage = message.get(0);
+			final Integer value = integerMessage.getValue();
+			if (value == 0) {
+				throw new PulsarBatchListenerFailedException("failed", integerMessage);
+			}
+			final Acknowledgement acknowledgment = invocation.getArgument(2);
+			List<MessageId> messageIds = new ArrayList<>();
+			for (Message<Integer> 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<Integer> container = new DefaultPulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
+
+		container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>(
+				new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10)));
+
+		container.start();
+
+		Map<String, Object> prodConfig = new HashMap<>();
+		prodConfig.put("topicName", "default-error-handler-tests-4");
+		final DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+				pulsarClient, prodConfig);
+		final PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+		for (int i = 0; i < 10; i++) {
+			pulsarTemplate.sendAsync(i);
+		}
+
+		PulsarOperations.SendMessageBuilder<Integer> 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<String, Object> 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<Integer> 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<Integer>> messages = invocation.getArgument(1);
+
+			for (Message<Integer> 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<Integer> container = new DefaultPulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
+
+		container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>(
+				new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10)));
+
+		container.start();
+
+		Map<String, Object> prodConfig = new HashMap<>();
+		prodConfig.put("topicName", "default-error-handler-tests-5");
+		final DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+				pulsarClient, prodConfig);
+		final PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+		for (int i = 0; i < 10; i++) {
+			pulsarTemplate.sendAsync(i);
+		}
+		PulsarOperations.SendMessageBuilder<Integer> 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<String, Object> 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<Integer> 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<Integer>> messages = invocation.getArgument(1);
+
+			for (Message<Integer> 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<Integer> container = new DefaultPulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
+
+		container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>(
+				new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10)));
+
+		container.start();
+
+		Map<String, Object> prodConfig = new HashMap<>();
+		prodConfig.put("topicName", "default-error-handler-tests-6");
+		final DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+				pulsarClient, prodConfig);
+		final PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+		for (int i = 0; i < 10; i++) {
+			pulsarTemplate.sendAsync(i);
+		}
+		PulsarOperations.SendMessageBuilder<Integer> 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<String, Object> 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<Integer> 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<Message<Integer>> messages = invocation.getArgument(1);
+			final Acknowledgement acknowledgment = invocation.getArgument(2);
+			for (Message<Integer> 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<Integer> container = new DefaultPulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
+
+		container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>(
+				new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10)));
+
+		container.start();
+
+		Map<String, Object> prodConfig = new HashMap<>();
+		prodConfig.put("topicName", "default-error-handler-tests-7");
+		final DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+				pulsarClient, prodConfig);
+		final PulsarTemplate<Integer> 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<String, Object> 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<Integer> 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<Message<Integer>> messages = invocation.getArgument(1);
+			final Acknowledgement acknowledgment = invocation.getArgument(2);
+			for (Message<Integer> 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<Integer> container = new DefaultPulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+		PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
+
+		container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>(
+				new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 10)));
+
+		container.start();
+
+		Map<String, Object> prodConfig = new HashMap<>();
+		prodConfig.put("topicName", "default-error-handler-tests-8");
+		final DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+				pulsarClient, prodConfig);
+		final PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+		for (int i = 0; i < 10; i++) {
+			pulsarTemplate.sendAsync(i);
+		}
+		PulsarOperations.SendMessageBuilder<Integer> 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();
+	}
+
+}