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

* List<Message<?>> was not casting properly for batch listeners.
  Addressing this issue.

spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java

@@ -222,6 +222,9 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
 		if (rawClass != null && isContainerType(rawClass)) {
 			resolvableType = resolvableType.getGeneric(0);
 		}
+		if (Message.class.isAssignableFrom(resolvableType.getRawClass())) {
+			resolvableType = resolvableType.getGeneric(0);
+		}
 		return resolvableType;
 	}
 

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

@@ -393,9 +393,18 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 		 * @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.");
+				AtomicBoolean messagesPendingInBatch, List<Message<T>> messageList, Throwable exception) {
+			try {
+				Assert.isInstanceOf(PulsarBatchListenerFailedException.class, exception,
+						"Batch listener should throw PulsarBatchListenerFailedException on errors.");
+			}
+			catch (Exception e) {
+				// try in the cause if something downstream wrapped the original
+				// exception.
+				exception = exception.getCause();
+				Assert.isInstanceOf(PulsarBatchListenerFailedException.class, exception,
+						"Batch listener should throw PulsarBatchListenerFailedException on errors.");
+			}
 			PulsarBatchListenerFailedException pulsarBatchListenerFailedException = (PulsarBatchListenerFailedException) exception;
 			Message<T> pulsarMessage = getPulsarMessageCausedTheException(pulsarBatchListenerFailedException);
 			final Message<T> theCurrentPulsarMessageTracked = this.pulsarConsumerErrorHandler.currentMessage();

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

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

spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java

@@ -82,7 +82,7 @@ public class PulsarBatchMessagingMessageListenerAdapter<V> extends PulsarMessagi
 			}
 		}
 		else {
-			message = null; // optimization since we won't need any conversion to invoke
+			message = MessageBuilder.withPayload(msg).build();
 		}
 		logger.debug(() -> "Processing [" + message + "]");
 

spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java

@@ -227,7 +227,8 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
 					&& parameterizedType.getActualTypeArguments().length == 1) {
 
 				Type paramType = parameterizedType.getActualTypeArguments()[0];
-				this.isConsumerRecordList = paramType.equals(Messages.class);
+				this.isConsumerRecordList = paramType instanceof ParameterizedType
+						&& ((ParameterizedType) paramType).getRawType().equals(Message.class);
 				boolean messageHasGeneric = paramType instanceof ParameterizedType && ((ParameterizedType) paramType)
 						.getRawType().equals(org.springframework.messaging.Message.class);
 				this.isMessageList = paramType.equals(org.springframework.messaging.Message.class) || messageHasGeneric;