Payload is not deserialized in SQS and SqsMessageConversionContext doesn't know a type of a POJO

[heroInCommunity]

Hey guys,

We have different SQS events on our project: each microservice has own format.
For instance, we may have event like
{ "sendTime": "2023-03-09T19:07:19.553Z", "messageBody": "{\"campaignId\":\"xxxx\",\"fromStatus\":\"DRAFT\",\"toStatus\":\"ENABLED\",\"changedAt\":\"2023-03-09T19:07:19.555Z\"}", "environment": "default", "build": "UNKNOWN", "cln": "UNKNOWN", "branch": "TRUNK", "correlationId": "fefewfew-fewfew3db4f93be" }

I would like to have an @SqsListener with a pojo which is serialized in "messageBody" field.
I tried to extend SqsMessagingMessageConverter by overriding getPayloadToDeserialize method.
As far as I understand, it can return either String or a POJO directly. However, I cannot convert to POJO in getPayloadToDeserialize because MessageConversionContext by some reason does not know a type of a POJO.

In version 2.x it was working fine (it was enough to create a custom MessageConverter which returned a string messageBody and using annotation io.awspring.cloud.messaging.config.annotation.NotificationMessage; in @SqsListener), however now in version 3.0.0.RC2 it does not work.

Could you please advice how could I implement the task described above?

[tomazfernandes]

Hey @heroInCommunity, thanks for bringing this up.

I never had thought of using getPayloadToDeserialize for this - it originally had a different intention. But turns out it's really an interesting extension point for your use case, thanks for pointing it out.

What you described worked for me - here's what I've done:

Mapped the full class

static class FullPayload {

  public String sendTime;
  public String messageBody;
  public String environment;
  public String build;
  public String branch;
  public String correlationId;

}

Created a custom SqsMessagingMessageConverter as you suggested

Now, as you mentioned, we can either return the serialized POJO as String or the POJO itself:

private static SqsMessagingMessageConverter createdNestedMessageConverter(ObjectMapper om) {
	return new SqsMessagingMessageConverter() {
		@Override
		protected Object getPayloadToDeserialize(software.amazon.awssdk.services.sqs.model.Message message) {
			try {
				FullPayload fullPayload = om.readValue(message.body(), FullPayload.class);
				// return fullPayload.messageBody;
				return om.readValue(fullPayload.messageBody, MyPojo.class);
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
		}
	};
}

Used MyPojo as the type in @SqsListener

@Component
static class ResolvesNestedPojoListener {

	@SqsListener(queueNames = RESOLVES_NESTED_POJO_QUEUE_NAME, id = "resolves-nested-pojo-mapping")
	void listen(MyPojo pojo, @Header(SqsHeaders.SQS_QUEUE_NAME_HEADER) String queueName) {
		Assert.isInstanceOf(MyPojo.class, pojo);
		logger.debug("Received message {} from queue {}", pojo, queueName);
	}
}

Worked like a charm both returning POJO and String.

Am I missing something?

Thanks.

[heroInCommunity]

Thank you for your reply. Indeed what you suggested will work, although SqsMessagingMessageConverter should know the POJO type.
In my case we have many listeners and we need to have a generic solution, so POJO type could not be hardcoded into any converter.
Using FullPayload is also not the best solution(because it would not be a generic one), since every event could have it's own set of unique fields around "messageBody".

The solution I mentioned (custom MessageConverter which returned a string messageBody and using annotation io.awspring.cloud.messaging.config.annotation.NotificationMessage; in @SqsListener) was working in a generic way in version 2.x

I am trying to find a generic solution for version 3.

The solution for version 2 was like

public class SqsMessageConverter implements MessageConverter {

  private final MappingJackson2MessageConverter mappingJackson2MessageConverter;

  @Override
  public Object fromMessage(Message<?> message, Class<?> targetClass) {
    Message<?> unwrappedMessage = unwrap(message, targetClass);
    return mappingJackson2MessageConverter.fromMessage(unwrappedMessage, targetClass);
  }

  private Message<?> unwrap(Message<?> message, Class<?> targetClass) {
    String unwrapped = String.valueOf(message.getPayload());

    if (targetClass.isAnnotationPresent(SqsMessageBody.class)) { // SqsMessageBody is our annotation which could be added to POJO event
      unwrapped = unwrap(unwrapped, "messageBody");
    } else if (targetClass.isAnnotationPresent(SqsMessage.class)) { // SqsMessage is our annotation which could be added to POJO event
      unwrapped = unwrap(unwrapped, "Message");
    }

    return new GenericMessage<>(unwrapped);
  }

  private String unwrap(String message, String property) {
    try {
      return Jackson.getObjectMapper()
                    .readTree(message)
                    .path(property)
                    .asText();
    } catch (IOException ex) {
      log.error("Failed to deserialize message {}", message, ex);
      return null;
    }
  }

  @Override
  public Message<?> toMessage(Object payload, MessageHeaders headers) {
    log.error("To message conversion is called for payload {}", payload);
    return null;
  }

}

With version 3 method fromMessage is not called anymore, because as I see from the code if there is no FQCN in an event body, then this method is not called.

and Listener like

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import org.springframework.cloud.aws.messaging.config.annotation.NotificationMessage;
import org.springframework.cloud.aws.messaging.listener.annotation.SqsListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@AllArgsConstructor
public class Listener {

  private final Processor processor;

  @SqsListener(value = "${config.sqs.listener}")
  public void onMessage(@NotificationMessage EventMessage message) {
    log.info("Receive event: {}", message);
    processor.process(message);
  }

}

[heroInCommunity]

Ok, finally I made it working by using combination

@AllArgsConstructor
public class MessageBodySqsMessagingMessageConverter extends SqsMessagingMessageConverter {

  private final ObjectMapper objectMapper;

  @Override
  protected HeaderMapper<Message> createDefaultHeaderMapper() {
    return new CorrelationIdSqsHeaderMapper(); // here I map correlationId to headers
  }

  @Override
  protected Object getPayloadToDeserialize(Message message) {
    String body = message.body();
    return unwrap(body, "messageBody");
  }

  private String unwrap(String message, String property) {
    try {
      JsonNode node = objectMapper.readTree(message);
      return node.path(property)
                 .asText();
    } catch (IOException ex) {
      return null;
    }
  }

}

and

@Slf4j
public class SqsMessageArgumentResolver implements HandlerMethodArgumentResolver {

  private final ObjectMapper objectMapper;

  public SqsMessageArgumentResolver(MappingJackson2MessageConverter mappingJackson2MessageConverter) {
    this.objectMapper = mappingJackson2MessageConverter.getObjectMapper();
  }

  @Override
  public boolean supportsParameter(MethodParameter parameter) {
    return ClassUtils.isAssignable(MessageWithCorrelationId.class, parameter.getParameterType());
  }

  @Override
  public MessageWithCorrelationId<Object> resolveArgument(MethodParameter parameter, Message<?> message) {
    Object payloadObject = message.getPayload();
    if (payloadObject instanceof String payload) {
      String correlationId = (String) message.getHeaders().get("correlationId");
      Object messageBody = getMessageBodyAsObject(payload, parameter);
      return new MessageWithCorrelationId<>(messageBody, correlationId);
    }
    log.error("Message payload is empty for message {}", message);
    return null;
  }

  private Object getMessageBodyAsObject(String messageBody, MethodParameter parameter) {
    Class<?> parametrizedType = getParametrizedType(parameter);
    try {
      return objectMapper.readValue(messageBody, parametrizedType);
    } catch (JsonProcessingException ex) {
      throw new RuntimeException(ex);
    }
  }

  private Class<?> getParametrizedType(MethodParameter parameter) {
    return (Class<?>) ((ParameterizedType) parameter.getGenericParameterType()).getActualTypeArguments()[0];
  }

}

it was important to add my argument resolver to the beginning of list of resolvers:

@Bean
  public SqsMessagingMessageConverter messagingMessageConverter() {
    MappingJackson2MessageConverter mappingJackson2MessageConverter = mappingJackson2MessageConverter();
    ObjectMapper objectMapper = mappingJackson2MessageConverter.getObjectMapper();
    SqsMessagingMessageConverter converter  = new MessageBodySqsMessagingMessageConverter(objectMapper);
    converter.setPayloadMessageConverter(mappingJackson2MessageConverter);
    return converter;
  }

  @Bean
  public SqsListenerConfigurer configurer(MappingJackson2MessageConverter messageConverter) {
    return registrar -> registerArgumentResolvers(messageConverter, registrar);
  }

  private void registerArgumentResolvers(MappingJackson2MessageConverter messageConverter,
                                         EndpointRegistrar registrar) {
    HandlerMethodArgumentResolver messageArgumentResolver = new SqsMessageArgumentResolver(messageConverter);
    registrar.manageMethodArgumentResolvers(list -> list.add(0, messageArgumentResolver));
  }

@Getter
@ToString
@NoArgsConstructor
@AllArgsConstructor
@JsonIgnoreProperties(ignoreUnknown = true)
public class MessageWithCorrelationId<T> {

  private T messageBody;
  private String correlationId;

}

I hope this could be useful for someone

[heroInCommunity]

for me getPayloadToDeserialize was never called

[tomazfernandes]

Would you mind giving this another shot with the solution above? I don't see why it wouldn't call this method, and it would be a cleaner solution than what you currently have.

If you can just run it and share what error it throws (if any) it'd be helpful. Thanks.

[heroInCommunity]

ok, I will try

[heroInCommunity]

No, it does not work for me, the only error I see is "Error processing message here-id-of-message", so it does not even entering the listener body

[tomazfernandes]

Hmm, that's odd. Would you mind sharing the stacktrace so we can see where it failed?

Thanks again!