SqsListener ignores custom MappingJackson2MessageConverter - bug or not?

[jkatnik]

I'm in the middle of a migration from Spring Boot 2.7 to 3.0.
It forces me to also migrate spring-cloud-aws from 2.4.2 to 2022.0.1

In 2.4 @SqsListener was utilizing the custom MappingJackson2MessageConverter bean created in our @Configuration class.

In 2022 it looks like spring-cloud-aws doesn't search for MappingJackson2MessageConverter in the application context but instead creates its own instance.

@Configuration
class JacksonConfig {
  @Bean
  @Primary
  fun objectMapper() = ObjectMapper()
      .findAndRegisterModules() // registers Kotlin Module
       ...

  @Bean
  @Primary
  fun mappingJackson2MessageConverter(objectMapper: ObjectMapper): MappingJackson2MessageConverter  {
    val messageConverter = MappingJackson2MessageConverter()
    messageConverter.objectMapper = objectMapper
    return messageConverter
  }
}

  @SqsListener("\${app.events.sqs}") // this listener doesn't use custom converter thus fails to deserialize to Kotlin data classes
  fun onEvent(serializedEvent: String) {
    ...
  }

I went through 3.0.0-RC1 docs but I didn't find any information that this behavior has changed.

Is it a bug or am I doing sth incorrectly?

[maciejwalkowiak]

You should be able to configure custom ObjectMapper instance in 3.0.

Look at docs here:

• Message Conversion and Payload Deserialization
• Global Configuration for @SqsListeners

Let me know if it worked for you!

[jkatnik]

What finally worked for me is:

@Bean
  fun objectMapper(): ObjectMapper {
    return ObjectMapper()
      .findAndRegisterModules()
      .enable(USE_BIG_DECIMAL_FOR_FLOATS)
      .disable(FAIL_ON_UNKNOWN_PROPERTIES)
      .disable(WRITE_DATES_AS_TIMESTAMPS)
  }

  @Bean
  fun mappingJackson2MessageConverter(objectMapper: ObjectMapper): MappingJackson2MessageConverter {
    val messageConverter = MappingJackson2MessageConverter()
    messageConverter.serializedPayloadClass = String::class.java
    messageConverter.isStrictContentTypeMatch = false
    messageConverter.objectMapper = objectMapper
    return messageConverter
  }
  
  @Bean
  fun defaultSqsListenerContainerFactory(
    sqsAsyncClient: SqsAsyncClient,
    jacksonConverter: MappingJackson2MessageConverter,
  ): SqsMessageListenerContainerFactory<String> {
    val converter = SqsMessagingMessageConverter()
    converter.payloadMessageConverter = jacksonConverter
    converter.setPayloadTypeHeader(APPLICATION_JSON_VALUE)
    val factory = SqsMessageListenerContainerFactory.builder<String>()
      .configure {
        it.messageConverter(converter)
        it.acknowledgementMode(ON_SUCCESS)
      }
      .sqsAsyncClient(sqsAsyncClient)
      .build()
    return factory
  }

without converter.setPayloadTypeHeader(APPLICATION_JSON_VALUE) listener was failing with java.lang.ClassCastException: class com.domain.invoices.event.InvoiceMetadataIngestedEvent cannot be cast to class [B (com.domain.invoices.event.InvoiceMetadataIngestedEvent is in unnamed module of loader 'app'; [B is in module java.base of loader 'bootstrap')

Takeaway points / RC feedback:

1. 3.0 in contrast 2.4 is more cumbersome (doesn't simply take MappingJackson2MessageConverter bean)
2. documentation doesn't mention:
   that

messageConverter.serializedPayloadClass = String::class.java
messageConverter.isStrictContentTypeMatch = false

is needed in MappingJackson2MessageConverter

and

converter.setPayloadTypeHeader(APPLICATION_JSON_VALUE)

[jkatnik]

Feedback continued:

Another problem I came across is that <T> SendResult<T> send(String queue, T payload)
when invoked with string payload, i.e. `sqsTempl.send("myQueue", """{ "foo": "bar" }""") was serializing body (i.e. escaping quotes).

It would be great if SqsTemplate had sendRawBody(String queueName, String body) method.

This syntax is quite useful in tests, when onde doesn't need/want/can have an object instance.

[tomazfernandes]

I see what I was missing now - if I add .header(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON) to my test it fails.

Opened this issue, if anyone wants to contribute a PR:
#1144

Thanks.

[tomazfernandes]

Hey @jkatnik, thanks for the feedback!

I think we can add MappingJackson2MessageConverter to auto-configuration, no worries there.

About the ClassCastException and related items: are you using Spring Dev Tools? There's an known issue with that.

About the SqsTemplate, it shouldn't be serializing a String payload, I'll take a look into that.

[tomazfernandes]

Hey @jkatnik, I've created a couple of issues to track the points you brought:
#721
#697

I'll get to those soon. In the meantime, do you have any other feedback to provide regarding SQS? Any other issues migrating to 3.0?

Thanks.

[tomazfernandes]

@jkatnik, I've created this test and it passes:

	@Test
	void shouldNotSerializeStringPayload() {
		String queue = "test-queue";
		String payload = """
			{ "foo": "bar" }
			""";
		SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
		GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
		given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
			.willReturn(CompletableFuture.completedFuture(urlResponse));
		SendMessageResponse response = SendMessageResponse.builder().messageId(UUID.randomUUID().toString()).build();
		given(mockClient.sendMessage(any(SendMessageRequest.class)))
			.willReturn(CompletableFuture.completedFuture(response));
		SqsOperations template = SqsTemplate.newSyncTemplate(mockClient);
		template.send(queue, payload);
		ArgumentCaptor<SendMessageRequest> captor = ArgumentCaptor.forClass(SendMessageRequest.class);
		then(mockClient).should().sendMessage(captor.capture());
		SendMessageRequest capturedRequest = captor.getValue();
		assertThat(capturedRequest.messageBody()).isEqualTo(payload);
	}

This tells me SqsTemplate is not changing anything on the payload. What am I missing? Perhaps that's on AWS SDK side?

[tomazfernandes]

Hi @mmaeller, I've tried to reproduce this issue with no success.

Would you be able to create a sample project on GitHub that reproduces this behavior?

Thanks

[mmaeller]

Hi @tomazfernandes, please have a look here.
I provided a test case that reproduces the issue. It looks like it depends on how the message is published to the queue.

[tomazfernandes]

Hey @mmaeller, thanks for the test case, I was able to track down the issue.

To give some context, there are two points in the execution flow we may deserialize the payload - when we receive the message and when we invoke the listener method.

When we send the message using SqsTemplate it adds type information in the headers, and with that we can convert the payload before sending the message to Interceptors or any component before message listener method.

When we send the message using the SqsAsyncClient, no such header is added, and we can only convert the message in the listener method invocation - any other component will receive the serialized payload as String.

The first conversion uses the messageConverter we set in ContainerOptions, while the second one uses the one we set in the EndpointRegistraro through SqsListenerCustomizer beans.

This logic has room for improvement - we should be able to tell the payload class without the header, but it's less than trivial in the Spring messaging structure.

In our autoconfiguration we're only setting the registrar one, and not the ContainerOptions one.

I think the simplest fix would be having some logic like this:

	@ConditionalOnMissingBean
	@Bean
	public SqsMessageListenerContainerFactory<Object> defaultSqsListenerContainerFactory(
			ObjectProvider<SqsAsyncClient> sqsAsyncClient, ObjectProvider<AsyncErrorHandler<Object>> asyncErrorHandler,
			ObjectProvider<ErrorHandler<Object>> errorHandler,
			ObjectProvider<AsyncMessageInterceptor<Object>> asyncInterceptors,
			ObjectProvider<MessageInterceptor<Object>> interceptors,
			ObjectProvider<ObjectMapper> objectMapperProvider
	) {

                        SqsMessageListenerContainerFactory<Object> factory = new SqsMessageListenerContainerFactory<>();
                        //...
                        objectMapperProvider.ifAvailable(mapper -> {
			    SqsMessagingMessageConverter converter = new SqsMessagingMessageConverter();
			    converter.setObjectMapper(mapper);
			    factory.configure(options -> options.messageConverter(converter));
                         });
		return factory;
	}

Would you like to contribute a PR with a fix and test? Thanks.

[mmaeller]

@tomazfernandes
I will take a look :)

[mmaeller]

#845
@tomazfernandes