Fix batch listener error handling

Updates the batch listener error handler code to handle
the case where there is only a single outstanding
message in the batch list when retries are expired.

Fixes #998

Signed-off-by: darshimo <[email protected]>

Author: darshimo

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

@@ -781,15 +781,8 @@ private List<Message<T>> invokeBatchListenerErrorHandler(AtomicBoolean inRetryMo
 				this.pulsarConsumerErrorHandler.recoverMessage(this.consumer, pulsarMessage,
 						pulsarBatchListenerFailedException);
 				handleAck(pulsarMessage, txn);
-				if (messageList.size() == 1) {
-					messagesPendingInBatch.set(false);
-				}
-				else {
-					messageList = messageList.subList(1, messageList.size());
-				}
-				if (!messageList.isEmpty()) {
-					messagesPendingInBatch.set(true);
-				}
+				messageList.remove(0);
+				messagesPendingInBatch.set(!messageList.isEmpty());
 				this.pulsarConsumerErrorHandler.clearMessage();
 			}
 			return messageList;