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