io.awspring.cloud.sqs.listener.source.AbstractPollingMessageSource","thread_name":"sdk-async-response-0-644","level_value":40000,"stack_trace":"java.lang.NullPointerException: payloadMessageConverter returned null payload

[sri2387]

Hi,
we are trying to listen form an SQS , but message converter is throwing nullpointer with the payload.
Here is the code :

@configuration
public class OrderEventMessagingConfiguration {

private static final Logger LOG = LoggerFactory.getLogger(OrderEventMessagingConfiguration.class);


@Value( "${cloud.aws.region.static}" )
private String region;

@Value( "${cloud.aws.credentials.access-key}" )
private String accessKey;

@Value( "${cloud.aws.credentials.secret-key}" )
private String secretKey;

@Bean
public SqsTemplate queueMessagingTemplate( SqsAsyncClient sqsAsyncClient )
{
    return SqsTemplate.newTemplate(sqsAsyncClient);
}

@Bean
public SqsMessageListenerContainerFactory<Object> defaultSqsListenerContainerFactory(SqsAsyncClient sqsAsyncClient,
                                                                                     ObjectMapper objectMapper) {

    LOG.info("Started Creating SqsMessageListenerContainerFactory:");
    SqsMessagingMessageConverter sqsMessagingMessageConverter = new SqsMessagingMessageConverter();
    MappingJackson2MessageConverter messageConverter = new MappingJackson2MessageConverter() ;

    messageConverter.setStrictContentTypeMatch(false);
    messageConverter.setObjectMapper(objectMapper);

    sqsMessagingMessageConverter.setPayloadMessageConverter(messageConverter);
    LOG.info("SqsMessageListenerContainerFactory setup complete.");

    return SqsMessageListenerContainerFactory
            .builder()
            .configure(options -> options
                    .messageConverter(sqsMessagingMessageConverter)
            )
            .sqsAsyncClient(sqsAsyncClient)
            .build();
}

@Bean
public SqsAsyncClient sqsAsyncClient()
{
    SqsAsyncClientBuilder sqsAsyncClientBuilder = SqsAsyncClient
            .builder()
            .region(Region.of(region));

    // If an AWS access key is provided then use the basic credentials, otherwise, the credentials
    // will be handled by the container. This allows SQS to work locally and as a containerized microservice.
    if ( StringUtils.isNotBlank( accessKey ) )
    {
        sqsAsyncClientBuilder.credentialsProvider( StaticCredentialsProvider.create(AwsBasicCredentials.create(accessKey, secretKey) ) );
    }

    return sqsAsyncClientBuilder.build();
}

}

[sri2387]

Exception we observe is:

I beleive the message is not able to convert with mapping converter, but it should atleast throw an error , I did write code to deserialize and add invlaid data for all wrong data, but it is not invoking my method. Code I refactored to correct data is :

public SqsMessageListenerContainerFactory defaultSqsListenerContainerFactory(SqsAsyncClient sqsAsyncClient,
ObjectMapper objectMapper) {

    LOG.info("Started Creating SqsMessageListenerContainerFactory:");
    SqsMessagingMessageConverter sqsMessagingMessageConverter = new SqsMessagingMessageConverter();
    MappingJackson2MessageConverter messageConverter = new MappingJackson2MessageConverter() {
        @Override
        protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
            LOG.debug("Entering convertFromInternal with payload: {}", message.getPayload());

            String originalPayload = new String((byte[]) message.getPayload());

            LOG.info("Original message payload: {}", originalPayload);

            // Pre-process the message payload to sanitize it
            String sanitizedPayload = originalPayload.replaceAll(
                    "com\\.commercetools\\.api\\.models\\.common\\.[^@]+@[0-9a-f]+",
                    "\"invalid_data\""
            );
            LOG.info("Sanitized message payload: {}", sanitizedPayload);

            // Rebuild the sanitized message
            Message<?> sanitizedMessage =
                    org.springframework.messaging.support.MessageBuilder.withPayload(sanitizedPayload.getBytes())
                            .copyHeaders(message.getHeaders())
                            .build();

            try {
                Object convertedObject = super.convertFromInternal(sanitizedMessage, targetClass, conversionHint);
                if (convertedObject == null) {
                    LOG.error("Converted object is null after sanitization. Payload: {}", sanitizedPayload);
                }
                return convertedObject;
            } catch (Exception e) {
                LOG.error("Failed to convert message: " + sanitizedMessage.getPayload(), e);
                throw e;
            }
        }
    };

    messageConverter.setStrictContentTypeMatch(false);
    messageConverter.setObjectMapper(objectMapper);

    sqsMessagingMessageConverter.setPayloadMessageConverter(messageConverter);
    LOG.info("SqsMessageListenerContainerFactory setup complete.");

    return SqsMessageListenerContainerFactory
            .builder()
            .configure(options -> options
                    .messageConverter(sqsMessagingMessageConverter)
            )
            .sqsAsyncClient(sqsAsyncClient)
            .build();
}

STILL THROWING SAME ERROR

log_type":"application","logger_name":"io.awspring.cloud.sqs.listener.source.AbstractPollingMessageSource","thread_name":"sdk-async-response-0-644","level_value":40000,"stack_trace":"java.lang.NullPointerException: payloadMessageConverter returned null payload\n\tat java.base/java.util.Objects.requireNonNull(Objects.java:259)\n\tat io.awspring.cloud.sqs.support.converter.AbstractMessagingMessageConverter.convertPayload(AbstractMessagingMessageConverter.java:172)\n\tat io.awspring.cloud.sqs.support.converter.AbstractMessagingMessageConverter.toMessagingMessage(AbstractMessagingMessageConverter.java:153)\n\tat io.awspring.cloud.sqs.listener.source.AbstractMessageConvertingMessageSource.convertMessage(AbstractMessageConvertingMessageSource.java:91)\n\tat java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)\n\tat java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1708)\n\tat java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)\n\tat java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)\n\tat java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921)\n\tat java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)\n\tat java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)\n\tat io.awspring.cloud.sqs.listener.source.AbstractMessageConvertingMessageSource.convertMessages(AbstractMessageConvertingMessageSource.java:85)\n\tat java.base/java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:646)\n\tat java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)\n\tat java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179)\n\tat software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncApiCallMetricCollectionStage.lambda$execute$0(AsyncApiCallMetricCollectionStage.java:58)\n\tat java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)\n\tat java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)\n\tat java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)\n\tat java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179)\n\tat software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncApiCallTimeoutTrackingStage.lambda$execute$2(AsyncApiCallTimeoutTrackingStage.java:69)\n\tat java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)\n\tat java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)\n\tat java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)\n\tat java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179)\n\tat software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncRetryableStage$RetryingExecutor.lambda$attemptExecute$1(AsyncRetryableStage.java:177)\n\tat java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)\n\tat java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)\n\tat java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)\n\tat java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179)\n\tat software.amazon.awssdk.core.internal.http.pipeline.stages.MakeAsyncHttpRequestStage.lambda$null$0(MakeAsyncHttpRequestStage.java:105)\n\tat java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)\n\tat java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)\n\tat java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)\n\tat java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179)\n\tat software.amazon.awssdk.core.internal.http.pipeline.stages.MakeAsyncHttpRequestStage.completeResponseFuture(MakeAsyncHttpRequestStage.java:238)\n\tat software.amazon.awssdk.core.internal.http.pipeline.stages.MakeAsyncHttpRequestStage.lambda$executeHttpRequest$3(MakeAsyncHttpRequestStage.java:163)\n\tat java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:934)\n\tat java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:911)\n\tat java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)\n\tat java.base/