From 852ea4483ab4e907cd56622db86279f4e6cab4e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9A=B0=EC=84=B1?= Date: Fri, 4 Sep 2026 12:24:01 +0900 Subject: [PATCH] fix: map SNS message attributes to SQS listener headers --- .../support/converter/SqsHeaderMapper.java | 54 ++++++++++++++++ .../converter/SqsHeaderMapperTests.java | 63 +++++++++++++++++++ 2 files changed, 117 insertions(+) diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapper.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapper.java index 484439d6c..ff279c0a1 100644 --- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapper.java +++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapper.java @@ -23,6 +23,7 @@ import io.awspring.cloud.sqs.support.converter.legacy.LegacyJackson2SqsMessagingMessageConverter; import java.nio.ByteBuffer; import java.time.Instant; +import java.util.Base64; import java.util.HashMap; import java.util.Map; import java.util.Objects; @@ -42,6 +43,10 @@ import software.amazon.awssdk.services.sqs.model.Message; import software.amazon.awssdk.services.sqs.model.MessageAttributeValue; import software.amazon.awssdk.services.sqs.model.MessageSystemAttributeName; +import tools.jackson.core.JacksonException; +import tools.jackson.core.type.TypeReference; +import tools.jackson.databind.JsonNode; +import tools.jackson.databind.json.JsonMapper; /** * A {@link HeaderMapper} implementation for SQS {@link Message}s. Enables creating additional SQS related headers from @@ -59,6 +64,11 @@ public class SqsHeaderMapper implements ContextAwareHeaderMapper { private static final Logger logger = LoggerFactory.getLogger(SqsHeaderMapper.class); + private static final TypeReference> SNS_MESSAGE_ATTRIBUTES_TYPE = new TypeReference<>() { + }; + + private final JsonMapper jsonMapper = new JsonMapper(); + private BiFunction additionalHeadersFunction = ((message, accessor) -> accessor.toMessageHeaders()); @@ -163,6 +173,7 @@ public MessageHeaders toHeaders(Message source) { accessor.copyHeadersIfAbsent(getMessageAttributesAsHeaders(source)); accessor.copyHeadersIfAbsent(createDefaultHeaders(source)); accessor.copyHeadersIfAbsent(createAdditionalHeaders(source)); + accessor.copyHeadersIfAbsent(getSnsMessageAttributesAsHeaders(source)); MessageHeaders messageHeaders = accessor.toMessageHeaders(); logger.trace("Mapped headers {} for message {}", messageHeaders, source.messageId()); @@ -191,6 +202,36 @@ private Map getMessageAttributesAsHeaders(Message source) { .collect(Collectors.toMap(Map.Entry::getKey, this::getValue)); } + private Map getSnsMessageAttributesAsHeaders(Message source) { + String body = source.body(); + if (body == null || !body.contains("\"MessageAttributes\"")) { + return Map.of(); + } + try { + JsonNode jsonNode = jsonMapper.readTree(body); + if (!isSnsNotification(jsonNode)) { + return Map.of(); + } + JsonNode messageAttributes = jsonNode.get("MessageAttributes"); + if (messageAttributes == null || !messageAttributes.isObject()) { + return Map.of(); + } + Map attributes = jsonMapper.convertValue(messageAttributes, + SNS_MESSAGE_ATTRIBUTES_TYPE); + return attributes.entrySet().stream() + .collect(Collectors.toMap(Map.Entry::getKey, entry -> getValue(entry.getValue()))); + } + catch (JacksonException | IllegalArgumentException e) { + logger.trace("Could not map SNS message attributes for message " + source.messageId(), e); + return Map.of(); + } + } + + private boolean isSnsNotification(JsonNode jsonNode) { + JsonNode type = jsonNode.get("Type"); + return type != null && "Notification".equals(type.asString()) && jsonNode.has("Message"); + } + private Object getValue(Map.Entry entry) { MessageAttributeValue value = entry.getValue(); String dataType = value.dataType(); @@ -207,6 +248,19 @@ private Object getValue(Map.Entry entry) { }; } + private Object getValue(SnsNotification.MessageAttribute value) { + String dataType = value.getType(); + Assert.notNull(dataType, "dataType must not be null"); + String baseDataType = dataType.contains(".") ? dataType.substring(0, dataType.indexOf('.')) : dataType; + + return switch (baseDataType) { + case MessageAttributeDataTypes.NUMBER -> getNumberValue(value.getValue(), dataType); + case MessageAttributeDataTypes.BINARY -> SdkBytes + .fromByteArray(Base64.getDecoder().decode(value.getValue())); + default -> value.getValue(); + }; + } + private Map getMessageSystemAttributesAsHeaders(Message source) { return source .attributes() diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapperTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapperTests.java index 05944615e..b0a32dca5 100644 --- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapperTests.java +++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapperTests.java @@ -144,6 +144,69 @@ void shouldAddNumberMessageAttributes() { assertThat(headers.get(headerName)).isEqualTo(headerValue); } + @Test + void shouldAddSnsMessageAttributes() { + SqsHeaderMapper mapper = new SqsHeaderMapper(); + Message message = Message.builder().body(""" + { + "Type": "Notification", + "Message": "payload", + "MessageAttributes": { + "stringAttribute": { "Type": "String", "Value": "myString" }, + "numberAttribute": { "Type": "Number.java.lang.Integer", "Value": "10" }, + "binaryAttribute": { "Type": "Binary", "Value": "bXlCaW5hcnk=" } + } + } + """).messageId(UUID.randomUUID().toString()).build(); + + MessageHeaders headers = mapper.toHeaders(message); + + assertThat(headers.get("stringAttribute")).isEqualTo("myString"); + assertThat(headers.get("numberAttribute")).isEqualTo(10); + assertThat(headers.get("binaryAttribute")).isEqualTo(SdkBytes.fromUtf8String("myBinary")); + } + + @Test + void shouldPreferSqsMessageAttributesOverSnsMessageAttributes() { + SqsHeaderMapper mapper = new SqsHeaderMapper(); + Message message = Message.builder().body(""" + { + "Type": "Notification", + "Message": "payload", + "MessageAttributes": { + "attribute": { "Type": "String", "Value": "snsValue" } + } + } + """) + .messageAttributes( + Map.of("attribute", + MessageAttributeValue.builder().dataType(MessageAttributeDataTypes.STRING) + .stringValue("sqsValue").build())) + .messageId(UUID.randomUUID().toString()).build(); + + MessageHeaders headers = mapper.toHeaders(message); + + assertThat(headers.get("attribute")).isEqualTo("sqsValue"); + } + + @Test + void shouldIgnoreMessageAttributesInNonSnsPayload() { + SqsHeaderMapper mapper = new SqsHeaderMapper(); + Message message = Message.builder().body(""" + { + "Type": "ApplicationEvent", + "Message": "payload", + "MessageAttributes": { + "attribute": { "Type": "String", "Value": "value" } + } + } + """).messageId(UUID.randomUUID().toString()).build(); + + MessageHeaders headers = mapper.toHeaders(message); + + assertThat(headers).doesNotContainKey("attribute"); + } + @Test void shouldCreateMessageWithSystemAttributesFromHeaders() { MessageHeaders headers = new MessageHeaders(