I am using Java 21 with Spring Boot 3 and Spring 6 to send a custom object to an SQS queue.
public class Event {
String eventName;
Instant dateTime;
}
My SQS message listener is set up as follows:
@SqsListener(value = "${sqs.queue.event.in}", acknowledgementMode = SqsListenerAcknowledgementMode.MANUAL)
public void receiveHourEvent(Acknowledgement acknowledgement, String event) {
///do something
}
To handle JSON serialization and deserialization properly, I have configured a custom ObjectMapper in my Spring configuration:
public class MyConfig {
@Bean
SqsMessageListenerContainerFactory<Object> defaultSqsListenerContainerFactory(SqsAsyncClient sqsAsyncClient,
SqsMessagingMessageConverter sqsMessagingMessageConverter,
SqsReceiverErrorHandler sqsReceiverErrorHandler,
@Value("${sqs.poll.timeout}") Duration pollTimeout,
@Value("${sqs.poll.messagevisibility}") Duration pollMessageVisibility) {
return SqsMessageListenerContainerFactory
.builder()
.configure(options -> options
.maxMessagesPerPoll(10)
.pollTimeout(pollTimeout)
.messageVisibility(pollMessageVisibility)
.messageConverter(sqsMessagingMessageConverter))
.sqsAsyncClient(sqsAsyncClient)
.messageInterceptor(new BreadcrumbMessageInterceptor())
.errorHandler(sqsReceiverErrorHandler)
.build();
}
@Bean
@Primary
public ObjectMapper objectMapper() {
ObjectMapper objectMapper = JsonMappers.getObjectMapper();
// Register the JavaTimeModule to support Java 8 date/time types
objectMapper.registerModule(new JavaTimeModule());
objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
return objectMapper;
}
@Bean
public SqsMessagingMessageConverter sqsMessagingMessageConverter(SqsExtendedClient sqsExtendedClient, ObjectMapper objectMapper) {
ExtendedSqsMessagingMessageConverter converter = new ExtendedSqsMessagingMessageConverter(sqsExtendedClient);
converter.setObjectMapper(objectMapper);
return converter;
}
}
@Bean
SqsTemplate sqsTemplate(SqsAsyncClient sqsAsyncClient, SqsMessagingMessageConverter sqsMessagingMessageConverter, MappingJackson2MessageConverter mappingJackson2MessageConverter) {
SqsTemplate sqsTemplate = SqsTemplate.builder()
.sqsAsyncClient(sqsAsyncClient)
.messageConverter(sqsMessagingMessageConverter)
.build();
return sqsTemplate;
}
I have ensured there are no dependency conflicts and that the required Jackson modules for Java 8 date/time support are included:
<!--JSON serialization-->
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
<version>2.17.2</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.17.2</version>
</dependency>
When sending a message to the queue:
sqsTemplate.send(queueName, event);
The message is sent as a custom object, but since my listener accepts a String, it fails to deserialize the Instant dateTime field properly.
Full error log:
.springframework.messaging.converter.MessageConversionException: Could not read JSON: Java 8 date/time type `java.time.Instant` not supported by default: add Module "com.fasterxml.jackson.datatype:jackson-datatype-jsr310" to enable handling
at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 122] (through reference chain: no.embriq.quant.flow.typelibmon.QFEvent["dateCreated"])
at .springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:256)
at .springframework.messaging.converter.AbstractMessageConverter.fromMessage(AbstractMessageConverter.java:183)
at .springframework.messaging.converter.AbstractMessageConverter.fromMessage(AbstractMessageConverter.java:174)
at .springframework.messaging.converter.CompositeMessageConverter.fromMessage(CompositeMessageConverter.java:57)
at io.awspring.cloud.sqs.support.converter.AbstractMessagingMessageConverter.convertPayload(AbstractMessagingMessageConverter.java:182)
at io.awspring.cloud.sqs.support.converter.AbstractMessagingMessageConverter.toMessagingMessage(AbstractMessagingMessageConverter.java:163)
at io.awspring.cloud.sqs.listener.source.AbstractMessageConvertingMessageSource.convertMessage(AbstractMessageConvertingMessageSource.java:99)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1708)
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921)
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)
at io.awspring.cloud.sqs.listener.source.AbstractMessageConvertingMessageSource.convertMessages(AbstractMessageConvertingMessageSource.java:90)
at java.base/java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:646)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFutureplete(CompletableFuture.java:2179)
at software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncApiCallMetricCollectionStage.lambda$execute$0(AsyncApiCallMetricCollectionStage.java:58)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFutureplete(CompletableFuture.java:2179)
at software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncApiCallTimeoutTrackingStage.lambda$execute$2(AsyncApiCallTimeoutTrackingStage.java:69)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFutureplete(CompletableFuture.java:2179)
at software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncRetryableStage$RetryingExecutor.lambda$attemptExecute$1(AsyncRetryableStage.java:177)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFutureplete(CompletableFuture.java:2179)
at software.amazon.awssdk.core.internal.http.pipeline.stages.MakeAsyncHttpRequestStage.lambda$execute$0(MakeAsyncHttpRequestStage.java:110)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFutureplete(CompletableFuture.java:2179)
at software.amazon.awssdk.core.internal.http.pipeline.stages.MakeAsyncHttpRequestStagepleteResponseFuture(MakeAsyncHttpRequestStage.java:253)
at software.amazon.awssdk.core.internal.http.pipeline.stages.MakeAsyncHttpRequestStage.lambda$executeHttpRequest$3(MakeAsyncHttpRequestStage.java:167)
at java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:934)
at java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:911)
at java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Java 8 date/time type `java.time.Instant` not supported by default: add Module "com.fasterxml.jackson.datatype:jackson-datatype-jsr310" to enable handling
at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 122] (through reference chain: no.embriq.quant.flow.typelibmon.QFEvent["dateCreated"])
at com.fasterxml.jackson.databind.exc.InvalidDefinitionException.from(InvalidDefinitionException.java:67)
at com.fasterxml.jackson.databind.DeserializationContext.reportBadDefinition(DeserializationContext.java:1887)
at com.fasterxml.jackson.databind.deser.impl.UnsupportedTypeDeserializer.deserialize(UnsupportedTypeDeserializer.java:48)
at com.fasterxml.jackson.databind.deser.impl.FieldProperty.deserializeAndSet(FieldProperty.java:138)
at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserializeFromObject(BeanDeserializer.java:399)
at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:185)
at com.fasterxml.jackson.databind.deser.DefaultDeserializationContext.readRootValue(DefaultDeserializationContext.java:342)
at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4905)
at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3848)
at .springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:251)
... 45 common frames omitted
How can I ensure that my custom ObjectMapper is correctly injected and used for deserialization in the listener?