Resolve listener payload type during container setup (#1548)

Resolve payload types during container initialization by analyzing
@SqsListener method signatures, removing the need for type headers by default.

With this change, the resolved payload type is available earlier in the flow,
and components such as MessageInterceptors, ErrorHandlers, and
AcknowledgementResultCallback receive the deserialized payload even when no
type header is present.

This removes coupling to producer-specific type names (which often differ across
services) and reduces risk from trusting type headers.

Supported patterns:
- Simple types: MyEvent
- Generic types: List<MyEvent>, Message<MyEvent>, List<Message<MyEvent>>
- Explicit @payload annotations

The SqsTemplate JavaType header is deprecated and ignored by default, with
documented options to restore the previous behavior. Polymorphic payloads
(interfaces, Object, @SqsHandler) require configuring a custom
payloadTypeMapper in SqsMessagingMessageConverter for type resolution.

Fixes #1546

Author: tomazfernandes

docs/src/main/asciidoc/sqs.adoc

@@ -660,6 +660,18 @@ public class MyListener {
 
 The `isDefault = true` parameter designates a method as the fallback handler for messages that don't match any other handler's parameter type.
 
+To determine which handler method to invoke, the framework needs to know the payload type before deserialization.
+A custom `PayloadTypeMapper` should be configured to map incoming messages to their concrete types.
+See <<Custom Payload Type Mapping>> for an example.
+
+[NOTE]
+====
+Since 4.0.0, the default type header sent by `SqsTemplate` is deprecated and is no longer used to deserialize the payload.
+A custom mapper is necessary to disambiguate between different payload types in multi-handler listeners.
+Alternatively, automatic inference can be disabled to restore header-based type resolution.
+See <<Automatic Payload Type Inference>> for more information.
+====
+
 ===== SNS Messages
 
 Since 3.1.1, when receiving SNS messages through the `@SqsListener`, the message includes all attributes of the `SnsNotification`. To only receive need the `Message` part of the payload, you can utilize the `@SnsNotificationMessage` annotation.
@@ -1609,16 +1621,55 @@ NOTE: When using Spring Boot's auto-configuration, if there's a single `ObjectMa
 This includes the one provided by Spring Boot's auto-configuration itself.
 For configuring a different `ObjectMapper`, see <<Global Configuration for @SqsListeners>>.
 
-For manually created `MessageListeners`, `MessageInterceptor` and `ErrorHandler` components, or more fine-grained conversion such as using `interfaces` or `inheritance` in listener methods, type mapping is required for payload deserialization.
+==== Automatic Payload Type Inference
+
+Since 4.0.0, by default, the framework automatically infers the payload type from the `@SqsListener` method signature at the `MessageSource` level.
+This allows payloads to be deserialized early in the message processing flow without requiring type information in message headers.
+
+This enables accessing the deserialized payload in components such as `MessageInterceptor`, `ErrorHandler`, and `AcknowledgementResultCallback` without type headers.
+
+The inference supports simple types, generic types like `List<MyEvent>`, `Message<MyEvent>`, and `List<Message<MyEvent>>`.
+Parameters annotated with `@Payload` are explicitly recognized as the payload parameter.
+
+For polymorphic types (interfaces, `Object`, or `@SqsHandler` methods), a custom mapper is required.
+See <<Custom Payload Type Mapping>>.
+
+[NOTE]
+====
+The `JavaType` header automatically set by `SqsTemplate` is deprecated since 4.0.0 and will be removed in a future version.
+When automatic payload type inference is enabled, the `JavaType` header is ignored by default.
+For backward compatibility, `SqsTemplate` still sends this header, allowing applications to roll back if needed.
+====
+
+==== Customizing Type Resolution
+
+Type resolution can be customized at two levels:
 
-By default, the framework looks for a `MessageHeader` named `Sqs_MA_JavaType` containing the fully qualified class name (`FQCN`) for which the payload should be deserialized to.
-If such header is found, the message is automatically deserialized to the provided class.
+*Per-converter:* Configure `setPayloadTypeMapper` or `setPayloadTypeHeader` on a `MessagingMessageConverter`.
+Any custom type mapper takes precedence over automatic inference for that converter.
+Note that, by default, all containers share the same SqsMessagingMessageConverter instance.
 
-Further configuration can be achieved by providing a configured `MessagingMessageConverter` instance in the `SqsContainerOptions`.
+*Framework-wide:* Provide a custom `MethodPayloadTypeInferrer` implementation via a `SqsListenerConfigurer`.
+See <<Global Configuration for @SqsListeners>>.
 
-NOTE: If type mapping is setup or type information is added to the headers, payloads are deserialized right after the message is polled.
-Otherwise, for `@SqsListener` annotated methods, payloads are deserialized right before the message is sent to the listener.
-For providing custom `MessageConverter` instances to be used by `@SqsListener` methods, see <<Global Configuration for @SqsListeners>>
+==== Disabling Automatic Inference
+
+*Per-converter:* To restore header-based type mapping for specific converters, call `setPayloadTypeHeader` on the converter:
+
+[source, java]
+----
+converter.setPayloadTypeHeader(SqsHeaders.SQS_DEFAULT_TYPE_HEADER);
+----
+
+*Framework-wide:* To disable automatic inference globally, set `methodPayloadTypeInferrer` to `null` via a `SqsListenerConfigurer`:
+
+[source, java]
+----
+@Bean
+SqsListenerConfigurer sqsListenerConfigurer() {
+    return registrar -> registrar.setMethodPayloadTypeInferrer(null);
+}
+----
 
 ==== Configuring a MessagingMessageConverter
 
@@ -1677,41 +1728,72 @@ messageConverter.setPayloadMessageConverter(payloadConverter);
 factory.configure(options -> options.messageConverter(messageConverter));
 ----
 
-==== Interfaces and Subclasses in Listener Methods
+==== Custom Payload Type Mapping
 
-Interfaces and subclasses can be used in `@SqsListener` annotated methods by configuring a `type mapper`:
+When determining the payload type based on message content is necessary, a custom `payloadTypeMapper` can be configured:
 
 [source, java]
 ----
-messageConverter.setPayloadTypeMapper(message -> {
-    String eventTypeHeader = message.getHeaders().get("myEventTypeHeader", String.class);
-    return "eventTypeA".equals(eventTypeHeader)
-        ? MyTypeA.class
-        : MyTypeB.class;
-});
+@Bean
+SqsMessagingMessageConverter messageConverter() {
+    var converter = new SqsMessagingMessageConverter();
+    converter.setPayloadTypeMapper(message -> {
+        String payload = (String) message.getPayload();
+        if (payload.contains("\"type\":\"OrderEvent\"")) {
+            return OrderEvent.class;
+        }
+        if (payload.contains("\"type\":\"PaymentEvent\"")) {
+            return PaymentEvent.class;
+        }
+        return null;
+    });
+    return converter;
+}
 ----
 
-And then, in the listener method:
+This enables using interfaces or subclasses in listener methods:
 
 [source, java]
 ----
-@SpringBootApplication
-public class SqsApplication {
+public interface DomainEvent {}
+public record OrderEvent(String orderId, String status) implements DomainEvent {}
+public record PaymentEvent(String paymentId, BigDecimal amount) implements DomainEvent {}
 
-    public static void main(String[] args) {
-        SpringApplication.run(SqsApplication.class, args);
+@SqsListener("events-queue")
+void handleEvent(DomainEvent event) {
+    // event will be the correct concrete type based on the mapper
+}
+----
+
+The same configuration enables routing to different `@SqsHandler` methods:
+
+[source, java]
+----
+@SqsListener("events-queue")
+public class EventListener {
+
+    @SqsHandler
+    void handle(OrderEvent event) {
+        // handles OrderEvent
     }
 
-    // Retrieve the converted payload
-    @SqsListener("myQueue")
-    public void listen(MyInterface message) {
-        System.out.println(message);
+    @SqsHandler
+    void handle(PaymentEvent event) {
+        // handles PaymentEvent
     }
+}
+----
 
-    // Or retrieve a Message with the converted payload
-    @SqsListener("myOtherQueue")
-    public void listen(Message<MyInterface> message) {
-        System.out.println(message);
+Or when using `Object` as the parameter type:
+
+[source, java]
+----
+@SqsListener("events-queue")
+void handleEvent(Object event) {
+    if (event instanceof OrderEvent orderEvent) {
+        // handle order
+    } else if (event instanceof PaymentEvent paymentEvent) {
+        // handle payment
     }
 }
 ----
@@ -1933,6 +2015,9 @@ By default, `StringMessageConverter`, `SimpleMessageConverter` and `MappingJacks
 
 - `manageArgumentResolvers` - gives access to the list of argument resolvers that will be used to resolve the listener method arguments.
 The order of resolvers is important - `PayloadMethodArgumentResolver` should generally be last since it's used as default.
+- `setMethodPayloadTypeInferrer` - set the `MethodPayloadTypeInferrer` instance to be used for automatic payload type inference.
+Set to `null` to disable automatic inference and rely on header-based type mapping.
+See <<Automatic Payload Type Inference>> for more information.
 
 A simple example would be:
 

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/ConfigUtils.java

@@ -92,4 +92,12 @@ public <T> ConfigUtils acceptMany(Collection<T> values, Consumer<T> consumer) {
 		return this;
 	}
 
+	public <T, V> ConfigUtils acceptManyIfNotNullAndInstance(@Nullable T value, Collection<?> values, Class<V> clazz,
+			BiConsumer<T, V> consumer) {
+		if (value != null) {
+			values.forEach(v -> acceptIfInstance(v, clazz, instance -> consumer.accept(value, instance)));
+		}
+		return this;
+	}
+
 }

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/annotation/AbstractListenerAnnotationBeanPostProcessor.java

@@ -99,6 +99,8 @@ public abstract class AbstractListenerAnnotationBeanPostProcessor<A extends Anno
 	@Nullable
 	private BeanExpressionContext expressionContext;
 
+	private final List<HandlerMethodArgumentResolver> argumentResolvers = new ArrayList<>();
+
 	@Override
 	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
 		Class<?> targetClass = AopUtils.getTargetClass(bean);
@@ -155,6 +157,7 @@ private Endpoint createAndConfigureEndpoint(Object bean, Method method, A annota
 			hme.setBean(bean);
 			hme.setMethod(method);
 			hme.setHandlerMethodFactory(this.delegatingHandlerMethodFactory);
+			hme.setArgumentResolvers(this.argumentResolvers);
 		});
 		return endpoint;
 	}
@@ -320,6 +323,7 @@ protected void configureDefaultHandlerMethodFactory(DefaultMessageHandlerMethodF
 		this.endpointRegistrar.getMethodArgumentResolversConsumer().accept(methodArgumentResolvers);
 		handlerMethodFactory.setArgumentResolvers(methodArgumentResolvers);
 		handlerMethodFactory.afterPropertiesSet();
+		this.argumentResolvers.addAll(methodArgumentResolvers);
 	}
 
 	protected Collection<HandlerMethodArgumentResolver> createAdditionalArgumentResolvers(

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/config/AbstractEndpoint.java

@@ -15,6 +15,7 @@
  */
 package io.awspring.cloud.sqs.config;
 
+import io.awspring.cloud.sqs.listener.AbstractMessageListenerContainer;
 import io.awspring.cloud.sqs.listener.AsyncMessageListener;
 import io.awspring.cloud.sqs.listener.BatchVisibility;
 import io.awspring.cloud.sqs.listener.ListenerMode;
@@ -23,6 +24,7 @@
 import io.awspring.cloud.sqs.listener.acknowledgement.BatchAcknowledgement;
 import io.awspring.cloud.sqs.listener.adapter.AsyncMessagingMessageListenerAdapter;
 import io.awspring.cloud.sqs.listener.adapter.MessagingMessageListenerAdapter;
+import io.awspring.cloud.sqs.support.converter.AbstractMessagingMessageConverter;
 import java.lang.reflect.Method;
 import java.util.Collection;
 import java.util.List;
@@ -35,6 +37,7 @@
 import org.springframework.core.MethodParameter;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
+import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
 import org.springframework.util.Assert;
 
@@ -59,6 +62,12 @@ public abstract class AbstractEndpoint implements HandlerMethodEndpoint {
 
 	private MessageHandlerMethodFactory handlerMethodFactory;
 
+	@Nullable
+	private MethodPayloadTypeInferrer methodPayloadTypeInferrer;
+
+	@Nullable
+	private List<HandlerMethodArgumentResolver> argumentResolvers;
+
 	protected AbstractEndpoint(Collection<String> queueNames, @Nullable String listenerContainerFactoryName,
 			String id) {
 		Assert.notEmpty(queueNames, "queueNames cannot be empty.");
@@ -127,6 +136,26 @@ public MessageHandlerMethodFactory getMessageHandlerMethodFactory() {
 		return this.handlerMethodFactory;
 	}
 
+	/**
+	 * Set the {@link MethodPayloadTypeInferrer} to be used for inferring payload types from method signatures.
+	 *
+	 * @param inferrer the inferrer instance, or null to disable inference.
+	 */
+	@Override
+	public void setMethodPayloadTypeInferrer(@Nullable MethodPayloadTypeInferrer inferrer) {
+		this.methodPayloadTypeInferrer = inferrer;
+	}
+
+	/**
+	 * Set the argument resolvers to be used for inferring payload types if a methodPayloadTypeInferrer is set.
+	 *
+	 * @param argumentResolvers the argument resolvers, may be null.
+	 */
+	@Override
+	public void setArgumentResolvers(@Nullable List<HandlerMethodArgumentResolver> argumentResolvers) {
+		this.argumentResolvers = argumentResolvers;
+	}
+
 	@Override
 	public void configureListenerMode(Consumer<ListenerMode> consumer) {
 		List<MethodParameter> parameters = getMethodParameters();
@@ -179,6 +208,17 @@ public <T> void setupContainer(MessageListenerContainer<T> container) {
 		Assert.notNull(this.handlerMethodFactory, "No handlerMethodFactory has been set");
 		InvocableHandlerMethod handlerMethod = this.handlerMethodFactory.createInvocableHandlerMethod(this.bean,
 				this.method);
+
+		if (this.methodPayloadTypeInferrer != null
+				&& container instanceof AbstractMessageListenerContainer<?, ?, ?> amlc) {
+			Class<?> inferredType = this.methodPayloadTypeInferrer.inferPayloadType(this.method,
+					this.argumentResolvers);
+			if (inferredType != null) {
+				amlc.setPayloadDeserializationType(inferredType);
+			}
+			disableDefaultPayloadTypeMapper(amlc);
+		}
+
 		if (CompletionStage.class.isAssignableFrom(handlerMethod.getReturnType().getParameterType())) {
 			container.setAsyncMessageListener(createAsyncMessageListenerInstance(handlerMethod));
 		}
@@ -187,6 +227,14 @@ public <T> void setupContainer(MessageListenerContainer<T> container) {
 		}
 	}
 
+	private void disableDefaultPayloadTypeMapper(AbstractMessageListenerContainer<?, ?, ?> container) {
+		var converter = container.getContainerOptions().getMessageConverter();
+		if (converter instanceof AbstractMessagingMessageConverter<?> amc && amc.isUsingDefaultPayloadTypeMapper()) {
+			// Disable JavaType header based mapping
+			amc.setPayloadTypeMapper(msg -> null);
+		}
+	}
+
 	protected <T> MessageListener<T> createMessageListenerInstance(InvocableHandlerMethod handlerMethod) {
 		return new MessagingMessageListenerAdapter<>(handlerMethod);
 	}