Is it necessary for the AbstractMessagingMessageConverter/SqsMessagingMessageConverter to send the JavaType header?

[kevinmcp123]

Hello 👋

I'll try circle back with a complete example when I've some more time but I'm seeing some odd behaviour with 3.1.X.

Basically I have a service that...

1. Consumes an SQS message using the @SqsListener annotation

	@SqsListener(value = "my-queue")
	public void receiveEvent( @Valid @NonNull final SnsMessage snsMessage ) {
		final OriginEvent event = deserialiseMessageField( snsMessage );  // Uses objectmapper readValue on sqsMessage.message
		this.ingestionService.process( event );
	}

2. Transforms OriginEvent to a new event (lets call that TxEvent)

3. Sends TxEvent using the SqsTemplate to a FIFO (standard autoconfigured template, no custom converter)

template.send( options -> options.queue( my-fifo.fifo )
						.payload( txEvent ) );

4. Tries to consume this event and map the json to RxEvent (almost identical to TxEvent but the json properties are mapped slightly different).

	@SqsListener(value = "my-fifo.fifo")
	public void receiveRxEvent( @Valid @NonNull final RxEvent event ) {
		this.ingestionService.process( event );
	}

5. Errors due to com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'RxEvent': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')

What I've observed is

• The MappingJackson2MessageConverter maps this from JSON to RxEvent successfully
• The debug trace ends up in MappingJackson2MessageConverter a second time (for some reason) and attempts to map between TxEvent & RxEvent.
• We see the error & fail

I think (but may be wrong) it does this due to the fact SqsMessagingMessageConverter adds a JavaType header for TxEvent.

---

Is this necessary? And why do we want to expose our package & class to the SQS consumer?

Thanks

N.b. This general approach seemed to work fine with 2.4.0

---

---

Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'RxEvent': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
 at [Source: (String)"RxEvent[...etc....]"; line: 1, column: 35]
	at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:2477) ~[jackson-core-2.15.3.jar:2.15.3]
	at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:760) ~[jackson-core-2.15.3.jar:2.15.3]
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._reportInvalidToken(ReaderBasedJsonParser.java:3041) ~[jackson-core-2.15.3.jar:2.15.3]
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._handleOddValue(ReaderBasedJsonParser.java:2082) ~[jackson-core-2.15.3.jar:2.15.3]
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:808) ~[jackson-core-2.15.3.jar:2.15.3]
	at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4912) ~[jackson-databind-2.15.3.jar:2.15.3]
	at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4818) ~[jackson-databind-2.15.3.jar:2.15.3]
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3772) ~[jackson-databind-2.15.3.jar:2.15.3]
	at org.springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:250) ~[spring-messaging-6.1.3.jar:6.1.3]
	... 26 common frames omitted

[kevvvvyp]

👋 I've raised an issue for this #1037 & also created a sample to replicate the issue (also linked on issue) https://github.com/kevvvvyp/spring-cloud-aws-sample

[tomazfernandes]

Hi @kevvvvyp, thanks for the issue and detailed explanation.

You're right in that the framework does deserialization in two places - the first is when the message is received, and the other is when the message gets to the listener method.

The reason for this is the listener method is dynamic, so it's not so simple to know beforehand the type the payload should be mapped to.

If the first deserialization succeeds, this means you will get a deserialized payload in components such as ErrorHandler and Message Interceptor, otherwise you'll get the JSON and only the proper POJO in the listener method.

This is a similar behavior to what happens in e.g. Spring for Apache Kafka.

If you're not interested in this, you can configure the Template to not add the type header.
https://docs.awspring.io/spring-cloud-aws/docs/3.0.0/reference/html/index.html#configuring-a-messagingmessageconverter

Let me know if that works for you.

[kevinmcp123]

Thanks @tomazfernandes 🙌 this worked a treat. I've updated my sample & will keep it for anyone else who has similar confusion https://github.com/kevvvvyp/spring-cloud-aws-sample.

I've went with the following config...

	@Bean
	public MessageConverter jackson2MessageConverter( final ObjectMapper objectMapper ) {
		final MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
		converter.setObjectMapper( objectMapper );
		converter.setSerializedPayloadClass( String.class );
		converter.setStrictContentTypeMatch( false );
		return converter;
	}

	@Bean
	public SqsMessagingMessageConverter sqsMessagingMessageConverter(
			final MessageConverter jackson2MessageConverter ) {
		// Create converter instance
		SqsMessagingMessageConverter messageConverter = new SqsMessagingMessageConverter();

		// Configure Type Header
		messageConverter.setPayloadTypeHeader( "payload-type-header" );

		// Configure Header Mapper
		SqsHeaderMapper headerMapper = new SqsHeaderMapper();
		headerMapper.setAdditionalHeadersFunction(
				( ( sqsMessage, accessor ) -> accessor.toMessageHeaders() ) );
		messageConverter.setHeaderMapper( headerMapper );

		// Configure Payload Converter
		messageConverter.setPayloadMessageConverter( jackson2MessageConverter );
		return messageConverter;
	}

	@Bean
	@Primary
	public SqsMessageListenerContainerFactory<Object> defaultSqsListenerContainerFactory(
			SqsMessagingMessageConverter messageConverter, SqsProperties sqsProperties,
			ObjectProvider<SqsAsyncClient> sqsAsyncClient,
			ObjectProvider<AsyncErrorHandler<Object>> asyncErrorHandler,
			ObjectProvider<ErrorHandler<Object>> errorHandler,
			ObjectProvider<AsyncMessageInterceptor<Object>> asyncInterceptors,
			ObjectProvider<MessageInterceptor<Object>> interceptors ) {

		SqsMessageListenerContainerFactory<Object> factory = new SqsMessageListenerContainerFactory<>();
		factory.configure( sqsContainerOptionsBuilder -> configureContainerOptions( sqsProperties,
				sqsContainerOptionsBuilder ) );
		sqsAsyncClient.ifAvailable( factory::setSqsAsyncClient );
		asyncErrorHandler.ifAvailable( factory::setErrorHandler );
		errorHandler.ifAvailable( factory::setErrorHandler );
		interceptors.forEach( factory::addMessageInterceptor );
		asyncInterceptors.forEach( factory::addMessageInterceptor );

		// Set MessageConverter to the factory or container
		factory.configure( options -> options.messageConverter( messageConverter ) );
		return factory;
	}

	private void configureContainerOptions( SqsProperties sqsProperties,
			ContainerOptionsBuilder<?, ?> options ) {
		PropertyMapper mapper = PropertyMapper.get().alwaysApplyingWhenNonNull();
		mapper.from( sqsProperties.getListener().getMaxConcurrentMessages() )
				.to( options::maxConcurrentMessages );
		mapper.from( sqsProperties.getListener().getMaxMessagesPerPoll() )
				.to( options::maxMessagesPerPoll );
		mapper.from( sqsProperties.getListener().getPollTimeout() ).to( options::pollTimeout );
	}

[tomazfernandes]

Sorry @kevinmcp123, I think I pointed you in the wrong direction.

If the goal is to not have the type header and hence skip the first deserialization, you can do this:

  @Bean
  public SqsTemplate sqsTemplate(SqsAsyncClient sqsAsyncClient) {
      return SqsTemplate.builder()
          .sqsAsyncClient(sqsAsyncClient)
          .configureDefaultConverter(converter -> converter.setPayloadTypeHeaderValueFunction(message -> null))
          .build();
  }

This should definitely be better documented - if you'd like to open a PR adding this to the docs it'd be great.

Thanks.

[kevind-wgu]

I have the same problem but want to prevent loading the class on the receiving side rather than the sending side. The receiver is a different application from the sender so the class does not exist in the same package. The SqsTemplate solution proposed above did not work for receiving messages.

To get around this issue on the receiving side I did the following.

@Slf4j
@Configuration
public class SqsMessageListenerOverrideConfiguration implements BeanPostProcessor {
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (bean instanceof SqsMessageListenerContainerFactory) {       
            SqsMessagingMessageConverter messageConverter = new SqsMessagingMessageConverter();
            messageConverter.setPayloadTypeMapper(m -> null); // Turn off finding the class from the Attribute Headers.
            ((SqsMessageListenerContainerFactory<Object>) bean).configure( options -> options.messageConverter( messageConverter ) );
        }
        return bean;
    }
}

@tomazfernandes Please let me know if there is a better way. (I Tried the SqsTemplate approach and it did not work for receiving messages).

[kevind-wgu]

@tomazfernandes to be honest I don't understand why the default deserialization tries to use the package/classname from the message to figure out the deserialization class. It seems like a bug waiting to happen. Especially if the application consuming the message is not the one that produced it. Additionally on the consuming method we are already required to specify the expected target object but the default deserialization code doesn't take this into consideration.

From a bug perspective say a developer decides to do some refactoring and moves the class to a different package or renames the class. This will cause deserialization to fail. But it is not obvious that this would happen.

Anyway, my preference would be to change this default behavior and change to let the handling method specify the expected object.

[gabrielevi10]

I agree with @kevind-wgu, I spent some hours today debugging until I found that the class the converter was trying to deserialize to was a different class from the one I set in the consumer method. Shouldn't the converter always try to deserialize to the class set in the consumer method first?

[tomazfernandes]

Yes, I agree with you both, and I don't like this solution very much either. Let me share some context on why this was done this way.

Even though the SqsMessageListenerContainer and its components were designed and implemented around AWS SDK v2 APIs especially for this project, the overall assembly-time structure of this project is largely drawn from Spring for Apache Kafka, and that's how the standard payload type resolution is handled there.

The reason this feature was used here was mainly because this way the payload is deserialized the moment the message is received from AWS SDK, and this way we get the deserialized POJO object inside the other components such as the Interceptors and ErrorHandler, before and after we invoke the listener method.

We could bring logic similar to the one used for type resolution from the listener to earlier in the flow, but it's not so simple since the listener method can have all kinds of argument types and resolvers, and we need to figure out which among these types corresponds to the payload. At the moment this project was being built, it was simpler to replicate this logic than going down that rabbit hole. But it's doable.

That being said, if we don't want to go down this road and implement this now, we could consider changing the default behavior not to have the deserialized payload in the other components, and making this an opt-in feature rather than opt-out, though this would be a reasonably big breaking change and likely something we don't want to do overnight.

Please let me know your thoughts, thanks.

[kevind-wgu]

The easiest solution would be to simply have a configuration item to easily turn it off. A better solution would be to make it the default setup so that people don't accidentally run into this situation like I did (fortunately it was while things were still in development). I can see how changing the default would be a major breaking change though. Ultimately I would prefer the longterm solution where the handler method payload type is figured out and used early in the SQS handling process.

[LeonPatmore]

+1 for supporting turning this off, it was quite confusing trying to figure this out (plus for many use-cases it doesn't work, as mentioned above)