From 91ea3ee2a3f452c8f9b822cdf40ad502837ad9e0 Mon Sep 17 00:00:00 2001 From: Marcos Tischer Vallim Date: Tue, 15 Sep 2026 14:22:01 -0300 Subject: [PATCH 1/6] fix: propagate throwable in `ResponseFailEntry` Signed-off-by: Marcos Tischer Vallim --- .../lib/core/AbstractAmazonSnsProducer.java | 5 +++- .../lib/core/AmazonSnsConsumerImpl.java | 22 +++++++++----- .../lib/core/AmazonSnsConsumerImpl.java | 29 ++++++++++++++----- 3 files changed, 41 insertions(+), 15 deletions(-) diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducer.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducer.java index ed9550a..6db8f3d 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducer.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducer.java @@ -74,11 +74,14 @@ public ListenableFuture send(final Requ } else { final ListenableFutureImpl listenableFutureImpl = new ListenableFutureImpl(Runnable::run); + final String message = String.format("Producer is currently in %s mode; no further messages will be accepted.", state.get().name()); + listenableFutureImpl.fail(ResponseFailEntry.builder() .withCode("000") .withId(requestEntry.getId()) - .withMessage(String.format("Producer is currently in %s mode; no further messages will be accepted.", state.get().name())) + .withMessage(message) .withSenderFault(true) + .withThrowable(new IllegalStateException(message)) .build() ); diff --git a/amazon-sns-java-messaging-lib-v1/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumerImpl.java b/amazon-sns-java-messaging-lib-v1/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumerImpl.java index 304052d..6d08ed8 100644 --- a/amazon-sns-java-messaging-lib-v1/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumerImpl.java +++ b/amazon-sns-java-messaging-lib-v1/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumerImpl.java @@ -36,6 +36,7 @@ import com.amazon.sns.messaging.lib.model.TopicProperty; import com.amazonaws.AmazonServiceException; import com.amazonaws.services.sns.AmazonSNS; +import com.amazonaws.services.sns.model.AmazonSNSException; import com.amazonaws.services.sns.model.PublishBatchRequest; import com.amazonaws.services.sns.model.PublishBatchRequestEntry; import com.amazonaws.services.sns.model.PublishBatchResult; @@ -143,13 +144,20 @@ public void handleResponse(final PublishBatchResult publishBatchResult) { ); publishBatchResult.getFailed().forEach(entry -> - Optional.ofNullable(pendingRequests.remove(entry.getId())).ifPresent(listenableFuture -> - listenableFuture.fail(ResponseFailEntry.builder() - .withId(entry.getId()) - .withCode(entry.getCode()) - .withMessage(entry.getMessage()) - .withSenderFault(entry.getSenderFault()) - .build()) + Optional.ofNullable(pendingRequests.remove(entry.getId())).ifPresent(listenableFuture -> { + final AmazonServiceException throwable = new AmazonSNSException(entry.getMessage()); + throwable.setErrorCode(entry.getCode()); + throwable.setErrorMessage(entry.getMessage()); + throwable.setServiceName("SNS"); + + listenableFuture.fail(ResponseFailEntry.builder() + .withId(entry.getId()) + .withCode(entry.getCode()) + .withMessage(entry.getMessage()) + .withSenderFault(entry.getSenderFault()) + .withThrowable(throwable) + .build()); + } ) ); } diff --git a/amazon-sns-java-messaging-lib-v2/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumerImpl.java b/amazon-sns-java-messaging-lib-v2/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumerImpl.java index f6bec8a..4c939a4 100644 --- a/amazon-sns-java-messaging-lib-v2/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumerImpl.java +++ b/amazon-sns-java-messaging-lib-v2/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumerImpl.java @@ -35,11 +35,13 @@ import com.amazon.sns.messaging.lib.model.ResponseSuccessEntry; import com.amazon.sns.messaging.lib.model.TopicProperty; +import software.amazon.awssdk.awscore.exception.AwsErrorDetails; import software.amazon.awssdk.awscore.exception.AwsServiceException; import software.amazon.awssdk.services.sns.SnsClient; import software.amazon.awssdk.services.sns.model.PublishBatchRequest; import software.amazon.awssdk.services.sns.model.PublishBatchRequestEntry; import software.amazon.awssdk.services.sns.model.PublishBatchResponse; +import software.amazon.awssdk.services.sns.model.SnsException; // @formatter:off /** @@ -145,13 +147,26 @@ public void handleResponse(final PublishBatchResponse publishBatchResult) { ); publishBatchResult.failed().forEach(entry -> - Optional.ofNullable(pendingRequests.remove(entry.id())).ifPresent(listenableFuture -> - listenableFuture.fail(ResponseFailEntry.builder() - .withId(entry.id()) - .withCode(entry.code()) - .withMessage(entry.message()) - .withSenderFault(entry.senderFault()) - .build()) + Optional.ofNullable(pendingRequests.remove(entry.id())).ifPresent(listenableFuture -> { + final AwsErrorDetails awsErrorDetails = AwsErrorDetails.builder() + .errorCode(entry.code()) + .errorMessage(entry.message()) + .serviceName("SNS") + .build(); + + final AwsServiceException throwable = SnsException.builder() + .awsErrorDetails(awsErrorDetails) + .message(entry.message()) + .build(); + + listenableFuture.fail(ResponseFailEntry.builder() + .withId(entry.id()) + .withCode(entry.code()) + .withMessage(entry.message()) + .withSenderFault(entry.senderFault()) + .withThrowable(throwable) + .build()); + } ) ); } From ab1aa67c231b984bb1b64db052d73a27114f2b25 Mon Sep 17 00:00:00 2001 From: Marcos Tischer Vallim Date: Tue, 15 Sep 2026 14:38:51 -0300 Subject: [PATCH 2/6] fix: propagate throwable in `ResponseFailEntry` Signed-off-by: Marcos Tischer Vallim --- .../messaging/lib/core/AbstractAmazonSnsConsumerTest.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumerTest.java b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumerTest.java index c62890d..2576c83 100644 --- a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumerTest.java +++ b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumerTest.java @@ -40,9 +40,11 @@ import java.util.Collections; import java.util.LinkedList; import java.util.List; +import java.util.Queue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; @@ -811,7 +813,7 @@ static class TestableAmazonSnsConsumer extends AbstractAmazonSnsConsumer publishedBatchSizes = Collections.synchronizedList(new LinkedList<>()); + private final Queue publishedBatchSizes = new ConcurrentLinkedQueue<>(); TestableAmazonSnsConsumer( final Object amazonSnsClient, @@ -873,7 +875,7 @@ void setThrowOnPublish(final boolean throwOnPublish) { } List getPublishedBatchSizes() { - return publishedBatchSizes; + return new LinkedList<>(publishedBatchSizes); } int getTotalPublishedEntries() { From 88db72538469edd2bcad374a586731b3eb421744 Mon Sep 17 00:00:00 2001 From: Marcos Tischer Vallim Date: Fri, 18 Sep 2026 13:06:56 -0300 Subject: [PATCH 3/6] fix: propagate throwable in `ResponseFailEntry` Signed-off-by: Marcos Tischer Vallim --- .../lib/core/AbstractMessageAttributes.java | 11 ++- .../lib/core/RequestEntryInternalFactory.java | 14 ++- .../core/AbstractMessageAttributesTest.java | 10 +- .../core/RequestEntryInternalFactoryTest.java | 26 ++--- .../messaging/lib/core/MessageAttributes.java | 2 +- .../lib/core/MessageAttributesTest.java | 2 +- .../messaging/lib/core/MessageAttributes.java | 2 +- .../lib/core/MessageAttributesTest.java | 2 +- pom.xml | 98 ++++++++++++++----- 9 files changed, 113 insertions(+), 54 deletions(-) diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributes.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributes.java index 04711cd..b5a3df5 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributes.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributes.java @@ -17,6 +17,7 @@ package com.amazon.sns.messaging.lib.core; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -44,21 +45,29 @@ abstract class AbstractMessageAttributes { */ protected static final String BINARY = "Binary"; + protected static final Integer BINARY_SIZE = BINARY.getBytes(StandardCharsets.UTF_8).length; + /** * Data type constant for string message attributes. */ protected static final String STRING = "String"; + protected static final Integer STRING_SIZE = STRING.getBytes(StandardCharsets.UTF_8).length; + /** * Data type constant for number message attributes. */ protected static final String NUMBER = "Number"; + protected static final Integer NUMBER_SIZE = NUMBER.getBytes(StandardCharsets.UTF_8).length; + /** * Data type constant for string array message attributes. */ protected static final String STRING_ARRAY = "String.Array"; + protected static final Integer STRING_ARRAY_SIZE = STRING_ARRAY.getBytes(StandardCharsets.UTF_8).length; + /** * Converts a map of message header entries to a map of SDK-specific message attribute values. * @@ -112,7 +121,7 @@ protected static String stringArray(final List values) { ); } - return "[ " + String.join(", ", collect) + " ]"; + return String.format("[%s]", String.join(",", collect)); } /** diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactory.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactory.java index 11eedcb..21a3a59 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactory.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactory.java @@ -21,6 +21,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.function.ToIntFunction; import java.util.stream.Collectors; import com.amazon.sns.messaging.lib.exception.PoisonRequestEntryException; @@ -178,12 +179,15 @@ static class MessageAttributesInternal extends AbstractMessageAttributes> computeSize = clazz -> + NUMBER.concat(".").concat(clazz.getName()).getBytes(StandardCharsets.UTF_8).length; + /** * {@inheritDoc} */ @Override public Integer getEnumMessageAttribute(final Enum value) { - return value.name().getBytes(StandardCharsets.UTF_8).length; + return STRING_SIZE + value.name().getBytes(StandardCharsets.UTF_8).length; } /** @@ -191,7 +195,7 @@ public Integer getEnumMessageAttribute(final Enum value) { */ @Override public Integer getStringMessageAttribute(final String value) { - return value.getBytes(StandardCharsets.UTF_8).length; + return STRING_SIZE + value.getBytes(StandardCharsets.UTF_8).length; } /** @@ -199,7 +203,7 @@ public Integer getStringMessageAttribute(final String value) { */ @Override public Integer getNumberMessageAttribute(final Number value) { - return value.toString().getBytes(StandardCharsets.UTF_8).length; + return value.toString().getBytes(StandardCharsets.UTF_8).length + computeSize.applyAsInt(value.getClass()); } /** @@ -207,7 +211,7 @@ public Integer getNumberMessageAttribute(final Number value) { */ @Override public Integer getBinaryMessageAttribute(final ByteBuffer value) { - return value.remaining(); + return BINARY_SIZE + value.remaining(); } /** @@ -215,7 +219,7 @@ public Integer getBinaryMessageAttribute(final ByteBuffer value) { */ @Override public Integer getStringArrayMessageAttribute(final List values) { - return stringArray(values).getBytes(StandardCharsets.UTF_8).length; + return STRING_ARRAY_SIZE + stringArray(values).getBytes(StandardCharsets.UTF_8).length; } } diff --git a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributesTest.java b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributesTest.java index fac6a8f..8300569 100644 --- a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributesTest.java +++ b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributesTest.java @@ -37,7 +37,7 @@ class AbstractMessageAttributesTest { @BeforeEach void setUp() { - messageAttributes = new AbstractMessageAttributes() { + messageAttributes = new AbstractMessageAttributes<>() { @Override protected String getEnumMessageAttribute(final Enum value) { @@ -161,7 +161,7 @@ void testStringArrayFormatsStringsWithQuotes() { final String result = AbstractMessageAttributes.stringArray(list); - assertThat(result, is("[ \"foo\", \"bar\" ]")); + assertThat(result, is("[\"foo\",\"bar\"]")); } @Test @@ -173,14 +173,14 @@ void testStringArrayFiltersOutNonStringElements() { final String result = AbstractMessageAttributes.stringArray(list); - assertThat(result, is("[ \"valid\", \"also-valid\" ]")); + assertThat(result, is("[\"valid\",\"also-valid\"]")); } @Test void testStringArrayReturnsEmptyBracketsForEmptyList() { final String result = AbstractMessageAttributes.stringArray(Collections.emptyList()); - assertThat(result, is("[ ]")); + assertThat(result, is("[]")); } @Test @@ -192,7 +192,7 @@ void testStringArrayReturnsEmptyBracketsWhenNoStringElementsPresent() { final String result = AbstractMessageAttributes.stringArray(list); - assertThat(result, is("[ ]")); + assertThat(result, is("[]")); } } \ No newline at end of file diff --git a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactoryTest.java b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactoryTest.java index c2e342b..7bbcb99 100644 --- a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactoryTest.java +++ b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactoryTest.java @@ -400,7 +400,7 @@ void testMessageAttributesSizeWithStringAttributeCountsKeyAndValue() { final Integer result = factory.messageAttributesSize(entry); - assertThat(result, equalTo(8)); + assertThat(result, equalTo(14)); } @Test @@ -413,7 +413,7 @@ void testMessageAttributesSizeWithMultipleAttributesSumsAll() { final Integer result = factory.messageAttributesSize(entry); - assertThat(result, equalTo(11)); + assertThat(result, equalTo(23)); } @Test @@ -437,7 +437,7 @@ void testMessageAttributesSizeWithEnumAttribute() { final Integer result = factory.messageAttributesSize(entry); - assertThat(result, equalTo(10)); + assertThat(result, equalTo(16)); } @Test @@ -450,7 +450,7 @@ void testMessageAttributesSizeWithBinaryAttribute() { final Integer result = factory.messageAttributesSize(entry); - assertThat(result, equalTo(7)); + assertThat(result, equalTo(13)); } @Test @@ -581,49 +581,49 @@ void testSingletonInstanceIsSameObject() { void testGetEnumMessageAttributeReturnsNameLength() { final Integer result = instance.getEnumMessageAttribute(SampleEnum.VALUE_ONE); - assertThat(result, equalTo("VALUE_ONE".length())); + assertThat(result, equalTo(15)); } @Test void testGetEnumMessageAttributeShortName() { final Integer result = instance.getEnumMessageAttribute(SampleEnum.A); - assertThat(result, equalTo(1)); + assertThat(result, equalTo(7)); } @Test void testGetStringMessageAttributeReturnsLength() { final Integer result = instance.getStringMessageAttribute("hello"); - assertThat(result, equalTo(5)); + assertThat(result, equalTo(11)); } @Test void testGetStringMessageAttributeEmptyString() { final Integer result = instance.getStringMessageAttribute(""); - assertThat(result, equalTo(0)); + assertThat(result, equalTo(6)); } @Test void testGetNumberMessageAttributeInteger() { final Integer result = instance.getNumberMessageAttribute(12345); - assertThat(result, equalTo(5)); + assertThat(result, equalTo(29)); } @Test void testGetNumberMessageAttributeFloat() { final Integer result = instance.getNumberMessageAttribute(3.14f); - assertThat(result, equalTo(String.valueOf(3.14f).length())); + assertThat(result, equalTo(26)); } @Test void testGetNumberMessageAttributeNegativeNumber() { final Integer result = instance.getNumberMessageAttribute(-99); - assertThat(result, equalTo(3)); + assertThat(result, equalTo(27)); } @Test @@ -633,7 +633,7 @@ void testGetBinaryMessageAttributeReturnsRemaining() { final Integer result = instance.getBinaryMessageAttribute(buffer); - assertThat(result, equalTo(3)); + assertThat(result, equalTo(9)); } @Test @@ -642,7 +642,7 @@ void testGetBinaryMessageAttributeEmptyBuffer() { final Integer result = instance.getBinaryMessageAttribute(buffer); - assertThat(result, equalTo(0)); + assertThat(result, equalTo(6)); } @Test diff --git a/amazon-sns-java-messaging-lib-v1/src/main/java/com/amazon/sns/messaging/lib/core/MessageAttributes.java b/amazon-sns-java-messaging-lib-v1/src/main/java/com/amazon/sns/messaging/lib/core/MessageAttributes.java index 462b194..21a3836 100644 --- a/amazon-sns-java-messaging-lib-v1/src/main/java/com/amazon/sns/messaging/lib/core/MessageAttributes.java +++ b/amazon-sns-java-messaging-lib-v1/src/main/java/com/amazon/sns/messaging/lib/core/MessageAttributes.java @@ -49,7 +49,7 @@ protected MessageAttributeValue getStringMessageAttribute(final String value) { */ @Override protected MessageAttributeValue getNumberMessageAttribute(final Number value) { - return new MessageAttributeValue().withDataType(NUMBER + "." + value.getClass().getName()).withStringValue(value.toString()); + return new MessageAttributeValue().withDataType(NUMBER.concat(".").concat(value.getClass().getName())).withStringValue(value.toString()); } /** diff --git a/amazon-sns-java-messaging-lib-v1/src/test/java/com/amazon/sns/messaging/lib/core/MessageAttributesTest.java b/amazon-sns-java-messaging-lib-v1/src/test/java/com/amazon/sns/messaging/lib/core/MessageAttributesTest.java index 54a8934..e043f85 100644 --- a/amazon-sns-java-messaging-lib-v1/src/test/java/com/amazon/sns/messaging/lib/core/MessageAttributesTest.java +++ b/amazon-sns-java-messaging-lib-v1/src/test/java/com/amazon/sns/messaging/lib/core/MessageAttributesTest.java @@ -89,7 +89,7 @@ void testSuccessStringArrayHeader() { assertThat(attributes.containsKey("stringArray"), is(true)); assertThat(attributes.get("stringArray").getDataType(), is("String.Array")); - assertThat(attributes.get("stringArray").getStringValue(), is("[ \"123\", \"456\" ]")); + assertThat(attributes.get("stringArray").getStringValue(), is("[\"123\",\"456\"]")); } @Test diff --git a/amazon-sns-java-messaging-lib-v2/src/main/java/com/amazon/sns/messaging/lib/core/MessageAttributes.java b/amazon-sns-java-messaging-lib-v2/src/main/java/com/amazon/sns/messaging/lib/core/MessageAttributes.java index 0f5cd14..f8a169d 100644 --- a/amazon-sns-java-messaging-lib-v2/src/main/java/com/amazon/sns/messaging/lib/core/MessageAttributes.java +++ b/amazon-sns-java-messaging-lib-v2/src/main/java/com/amazon/sns/messaging/lib/core/MessageAttributes.java @@ -50,7 +50,7 @@ protected MessageAttributeValue getStringMessageAttribute(final String value) { */ @Override protected MessageAttributeValue getNumberMessageAttribute(final Number value) { - return MessageAttributeValue.builder().dataType(NUMBER + "." + value.getClass().getName()).stringValue(value.toString()).build(); + return MessageAttributeValue.builder().dataType(NUMBER.concat(".").concat(value.getClass().getName())).stringValue(value.toString()).build(); } /** diff --git a/amazon-sns-java-messaging-lib-v2/src/test/java/com/amazon/sns/messaging/lib/core/MessageAttributesTest.java b/amazon-sns-java-messaging-lib-v2/src/test/java/com/amazon/sns/messaging/lib/core/MessageAttributesTest.java index dd91e79..11c6cb3 100644 --- a/amazon-sns-java-messaging-lib-v2/src/test/java/com/amazon/sns/messaging/lib/core/MessageAttributesTest.java +++ b/amazon-sns-java-messaging-lib-v2/src/test/java/com/amazon/sns/messaging/lib/core/MessageAttributesTest.java @@ -90,7 +90,7 @@ void testSuccessStringArrayHeader() { assertThat(attributes.containsKey("stringArray"), is(true)); assertThat(attributes.get("stringArray").dataType(), is("String.Array")); - assertThat(attributes.get("stringArray").stringValue(), is("[ \"123\", \"456\" ]")); + assertThat(attributes.get("stringArray").stringValue(), is("[\"123\",\"456\"]")); } @Test diff --git a/pom.xml b/pom.xml index 88871d1..bdfadc1 100644 --- a/pom.xml +++ b/pom.xml @@ -25,7 +25,6 @@ 1.18.42 1.3.14 - 4.11.0 5.10.2 2.16.1 4.3.0 @@ -101,16 +100,19 @@ amazon-sns-java-messaging-lib-template ${project.version} + com.github.mvallim amazon-sns-java-messaging-lib-v1 ${project.version} + com.github.mvallim amazon-sns-java-messaging-lib-v2 ${project.version} + com.fasterxml.jackson jackson-bom @@ -118,40 +120,54 @@ pom import + + + org.mockito + mockito-bom + ${mockito.version} + pom + import + + + + io.micrometer + micrometer-bom + ${micrometer.version} + pom + import + + + + org.testcontainers + testcontainers-bom + ${testcontainers.version} + pom + import + + org.apache.fory fory-json ${fory.version} + org.slf4j slf4j-api ${slf4j-api.version} + org.apache.commons commons-collections4 ${commons-collections4.version} + org.apache.commons commons-lang3 ${commons-lang3.version} - - io.micrometer - micrometer-bom - ${micrometer.version} - pom - import - - - org.testcontainers - testcontainers-bom - ${testcontainers.version} - pom - import - @@ -231,21 +247,12 @@ org.mockito mockito-junit-jupiter - ${mockito.version} test org.mockito mockito-core - ${mockito.version} - test - - - - org.mockito - mockito-inline - ${mockito.version} test @@ -372,6 +379,17 @@ + + org.apache.maven.plugins + maven-dependency-plugin + + + + properties + + + + org.apache.maven.plugins maven-clean-plugin @@ -398,7 +416,7 @@ org.apache.maven.plugins maven-surefire-plugin - ${jacoco.argLine} -Dfile.encoding=${project.build.sourceEncoding} -Xms1024m -Xmx1024m -Dnet.bytebuddy.experimental=true + ${jacoco.argLine} ${mockito.argLine} -Dfile.encoding=${project.build.sourceEncoding} -Xms1024m -Xmx1024m @@ -467,51 +485,79 @@ 8 + 1.8 + 8 + 4.11.0 + + + + + org.mockito + mockito-inline + test + + 11 + 11 + 11 + 5.23.0 + -XX:+EnableDynamicAgentLoading -Xshare:off 17 + 17 + 17 + 5.23.0 + -XX:+EnableDynamicAgentLoading -Xshare:off 21 + 21 + 21 - + 5.23.0 + -XX:+EnableDynamicAgentLoading -Xshare:off + 25 + 25 + 25 + 5.23.0 + -XX:+EnableDynamicAgentLoading -Xshare:off From 5becf8b8e1d5c0d1c92e8d9016bb0a76470bcbd2 Mon Sep 17 00:00:00 2001 From: Marcos Tischer Vallim Date: Sun, 20 Sep 2026 17:19:04 -0300 Subject: [PATCH 4/6] fix: propagate throwable in `ResponseFailEntry` Signed-off-by: Marcos Tischer Vallim --- .../PoisonRequestEntryException.java | 30 +++++++++++++++---- .../lib/core/JsonMapperFactoryTest.java | 28 +++++++++++++++-- pom.xml | 11 +++++-- 3 files changed, 57 insertions(+), 12 deletions(-) diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/exception/PoisonRequestEntryException.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/exception/PoisonRequestEntryException.java index f7ec973..8bb6af4 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/exception/PoisonRequestEntryException.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/exception/PoisonRequestEntryException.java @@ -1,9 +1,26 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + package com.amazon.sns.messaging.lib.exception; /** - * Base exception for request entries that cannot be published and must be dropped. - * A "poison" request entry is one that fails serialization or exceeds the maximum - * allowed message size, making it impossible to include in a publish batch. + * Base exception for request entries that cannot be published and must be + * dropped. A "poison" request entry is one that fails serialization or exceeds + * the maximum allowed message size, making it impossible to include in a + * publish batch. *

* Concrete subclasses ({@link MaximumAllowedMessageException}, * {@link JsonProcessingException}) are created via the factory methods @@ -34,8 +51,8 @@ protected PoisonRequestEntryException(final String message, final Throwable caus } /** - * Creates a {@link MaximumAllowedMessageException} for a message that exceeds the - * maximum allowed size. + * Creates a {@link MaximumAllowedMessageException} for a message that exceeds + * the maximum allowed size. * * @param message the detail message * @return a new {@link MaximumAllowedMessageException} @@ -45,7 +62,8 @@ public static PoisonRequestEntryException fromMaximumAllowedMessage(final String } /** - * Creates a {@link JsonProcessingException} for a payload that failed to serialize. + * Creates a {@link JsonProcessingException} for a payload that failed to + * serialize. * * @param message the detail message * @param throwable the underlying serialization exception diff --git a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/JsonMapperFactoryTest.java b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/JsonMapperFactoryTest.java index 9111598..ebafb87 100644 --- a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/JsonMapperFactoryTest.java +++ b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/JsonMapperFactoryTest.java @@ -1,3 +1,19 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + package com.amazon.sns.messaging.lib.core; import static org.hamcrest.MatcherAssert.assertThat; @@ -100,7 +116,9 @@ void testToJsonDelegatesAndReturnsResult() throws Exception { @Test void testToJsonDelegateThrowsWrapsInPoisonRequestEntryException() throws Exception { final IOException cause = new IOException("boom"); - when(objectMapper.writeValueAsString(any())).thenAnswer(invocation -> { throw cause; }); + when(objectMapper.writeValueAsString(any())).thenAnswer(invocation -> { + throw cause; + }); final PoisonRequestEntryException ex = assertThrows(PoisonRequestEntryException.class, () -> jsonMapper.toJson(SAMPLE_VALUE)); @@ -120,7 +138,9 @@ void testToJsonBytesDelegatesAndReturnsResult() throws Exception { @Test void testToJsonBytesDelegateThrowsWrapsInPoisonRequestEntryException() throws Exception { final IOException cause = new IOException("boom"); - when(objectMapper.writeValueAsBytes(any())).thenAnswer(invocation -> { throw cause; }); + when(objectMapper.writeValueAsBytes(any())).thenAnswer(invocation -> { + throw cause; + }); final PoisonRequestEntryException ex = assertThrows(PoisonRequestEntryException.class, () -> jsonMapper.toJsonBytes(SAMPLE_VALUE)); @@ -140,7 +160,9 @@ void testFromJsonStringDelegatesAndReturnsResult() throws Exception { @Test void testFromJsonStringDelegateThrowsWrapsInPoisonRequestEntryException() throws Exception { final IOException cause = new IOException("malformed"); - when(objectMapper.readValue(any(String.class), eq(String.class))).thenAnswer(invocation -> { throw cause; }); + when(objectMapper.readValue(any(String.class), eq(String.class))).thenAnswer(invocation -> { + throw cause; + }); final PoisonRequestEntryException ex = assertThrows(PoisonRequestEntryException.class, () -> jsonMapper.fromJson(SAMPLE_JSON, String.class)); diff --git a/pom.xml b/pom.xml index bdfadc1..4cfdbd1 100644 --- a/pom.xml +++ b/pom.xml @@ -121,6 +121,14 @@ import + + org.junit + junit-bom + ${junit.jupiter.version} + pom + import + + org.mockito mockito-bom @@ -219,21 +227,18 @@ org.junit.jupiter junit-jupiter-engine - ${junit.jupiter.version} test org.junit.jupiter junit-jupiter-api - ${junit.jupiter.version} test org.junit.jupiter junit-jupiter-params - ${junit.jupiter.version} test From 4ac8ffd13e08997f91e1cb2e25b28927a6b1c440 Mon Sep 17 00:00:00 2001 From: Marcos Tischer Vallim Date: Sun, 20 Sep 2026 17:21:12 -0300 Subject: [PATCH 5/6] fix: propagate throwable in `ResponseFailEntry` Signed-off-by: Marcos Tischer Vallim --- .../sns/messaging/lib/core/AbstractMessageAttributesTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributesTest.java b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributesTest.java index 8300569..dc29586 100644 --- a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributesTest.java +++ b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractMessageAttributesTest.java @@ -37,7 +37,7 @@ class AbstractMessageAttributesTest { @BeforeEach void setUp() { - messageAttributes = new AbstractMessageAttributes<>() { + messageAttributes = new AbstractMessageAttributes() { @Override protected String getEnumMessageAttribute(final Enum value) { From 0ed8a0cbf66be14f9e337fa4aee8c04ebbf78ef4 Mon Sep 17 00:00:00 2001 From: Marcos Tischer Vallim Date: Thu, 24 Sep 2026 11:52:24 -0300 Subject: [PATCH 6/6] feat(max-message-size): flexible configuration to max message size Signed-off-by: Marcos Tischer Vallim --- .../lib/core/AbstractAmazonSnsConsumer.java | 16 +++----- .../messaging/lib/model/TopicProperty.java | 40 ++++++++++++++++++- .../core/AbstractAmazonSnsConsumerTest.java | 21 +++++----- 3 files changed, 54 insertions(+), 23 deletions(-) diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumer.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumer.java index 4f85363..9621190 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumer.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumer.java @@ -68,11 +68,6 @@ abstract class AbstractAmazonSnsConsumer implements Runnable, Amazon */ private static final Integer KB = 1024; - /** - * Maximum batch size threshold of 256 KB imposed by Amazon SNS. - */ - private static final Integer BATCH_SIZE_BYTES_THRESHOLD = 256 * AbstractAmazonSnsConsumer.KB; - /** Class logger. */ private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAmazonSnsConsumer.class); @@ -123,7 +118,7 @@ protected AbstractAmazonSnsConsumer( this.topicProperty = Objects.requireNonNull(topicProperty, "topicProperty cannot be null"); this.amazonSnsClient = Objects.requireNonNull(amazonSnsClient, "amazonSnsClient cannot be null"); this.executorService = Objects.requireNonNull(executorService, "executorService cannot be null"); - this.requestEntryInternalFactory = RequestEntryInternalFactory.build(jsonMapper); + requestEntryInternalFactory = RequestEntryInternalFactory.build(jsonMapper); this.pendingRequests = pendingRequests; this.topicRequests = topicRequests; this.publishDecorator = publishDecorator; @@ -271,7 +266,7 @@ private boolean maxBatchSizeReached(final BlockingQueue> request * @return true if the request can be added */ private boolean canAddToBatch(final int batchSizeBytes, final int requestEntriesSize, final RequestEntry request) { - return (batchSizeBytes < BATCH_SIZE_BYTES_THRESHOLD) + return (batchSizeBytes < topicProperty.getMaxMessageSize()) && (requestEntriesSize < topicProperty.getMaxBatchSize()) && Objects.nonNull(request); } @@ -283,7 +278,7 @@ private boolean canAddToBatch(final int batchSizeBytes, final int requestEntries * @return true if the batch is still within the size limit */ private boolean canAddPayload(final int batchSizeBytes) { - return batchSizeBytes <= BATCH_SIZE_BYTES_THRESHOLD; + return batchSizeBytes <= topicProperty.getMaxMessageSize(); } /** @@ -309,8 +304,9 @@ private Optional createBatch(final BlockingQueue> requests) t final Integer messageSize = messageBodySize + messageAttributesSize; - if (messageSize > BATCH_SIZE_BYTES_THRESHOLD) { - throw PoisonRequestEntryException.fromMaximumAllowedMessage("The maximum allowed message size exceeding 256KB (262,144 bytes)."); + if (messageSize > topicProperty.getMaxMessageSize()) { + final String message = String.format("The maximum allowed message size exceeding %dKB (%,d bytes).", topicProperty.getMaxMessageSize() / KB, topicProperty.getMaxMessageSize()); + throw PoisonRequestEntryException.fromMaximumAllowedMessage(message); } if (canAddPayload(batchSizeBytes.get() + messageSize)) { diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/model/TopicProperty.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/model/TopicProperty.java index 32afe36..bda0353 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/model/TopicProperty.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/model/TopicProperty.java @@ -51,8 +51,16 @@ @AllArgsConstructor(access = AccessLevel.PRIVATE) public class TopicProperty { + private static final int KB = 1024; + private static final long DEFAULT_LINGER = 10L; + private static final int DEFAULT_MESSAGE_SIZE = 256 * KB; + + private static final int MIN_MESSAGE_SIZE = KB; + + private static final int MAX_MESSAGE_SIZE = 1024 * KB; + /** * Whether the topic is a FIFO topic. */ @@ -78,6 +86,11 @@ public class TopicProperty { */ private final int maxBatchSize; + /** + * The maximum message size in bytes. + */ + private final int maxMessageSize; + @NoArgsConstructor(access = AccessLevel.PRIVATE) static final class TopicPropertyValidator extends AbstractValidator { @@ -112,6 +125,10 @@ public void rules() { .must(greaterThanOrEqual(DEFAULT_LINGER)) .withMessage("'linger' must be greater than or equal to 10 (ten)"); + ruleFor("maxMessageSize", TopicProperty::getMaxMessageSize) + .must(betweenInclusive(MIN_MESSAGE_SIZE, MAX_MESSAGE_SIZE)) + .withMessage("'maxMessageSize' must be in the range of 1Kb (1,024 bytes) to 1024Kb (1,048,576 bytes)"); + ruleFor("maxBatchSize", TopicProperty::getMaxBatchSize) .must(betweenInclusive(1, 10)) .withMessage("'maxBatchSize' must be in the range of 1 (one) to 10 (ten)"); @@ -151,6 +168,14 @@ public static class TopicPropertyBuilder { */ private boolean linger$set; + /** + * Tracks whether {@code maxMessageSize(int)} was explicitly invoked. + * + *

This flag allows applying {@link TopicProperty#DEFAULT_MESSAGE_SIZE} only when + * no explicit value was provided through the builder. + */ + private boolean maxMessageSize$set; + /** * Sets the batching linger time in milliseconds. * @@ -163,6 +188,18 @@ public TopicPropertyBuilder linger(final long linger) { return this; } + /** + * Sets the maximum message size in bytes. + * + * @param maxMessageSize the maximum message size in bytes + * @return this builder + */ + public TopicPropertyBuilder maxMessageSize(final int maxMessageSize) { + this.maxMessageSize = maxMessageSize; + maxMessageSize$set = true; + return this; + } + /** * Validates and builds the {@link TopicProperty}. If no linger value was set * explicitly, the default of 10 ms is applied. Throws @@ -173,8 +210,9 @@ public TopicPropertyBuilder linger(final long linger) { */ public TopicProperty build() { final long linger = linger$set ? this.linger : DEFAULT_LINGER; + final int maxMessageSize = maxMessageSize$set ? this.maxMessageSize : DEFAULT_MESSAGE_SIZE; - final TopicProperty topicProperty = new TopicProperty(fifo, maximumPoolSize, topicArn, linger, maxBatchSize); + final TopicProperty topicProperty = new TopicProperty(fifo, maximumPoolSize, topicArn, linger, maxBatchSize, maxMessageSize); final ValidationResult validationResult = TopicPropertyValidator.INSTANCE.validate(topicProperty); diff --git a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumerTest.java b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumerTest.java index 2576c83..420d754 100644 --- a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumerTest.java +++ b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumerTest.java @@ -117,6 +117,7 @@ void setUp() { when(topicProperty.getLinger()).thenReturn(LINGER_MS); when(topicProperty.getMaxBatchSize()).thenReturn(MAX_BATCH_SIZE); when(topicProperty.isFifo()).thenReturn(false); + when(topicProperty.getMaxMessageSize()).thenReturn(256 * TestableAmazonSnsConsumer.KB); } @Nested @@ -629,7 +630,7 @@ void testCanAddPayloadAllowsEntryExactlyAtSizeThreshold() throws Exception { when(topicProperty.isFifo()).thenReturn(true); context(consumer -> { - final String payloadAtThreshold = buildPayloadOfBytes(TestableAmazonSnsConsumer.batchSizeBytesThreshold()); + final String payloadAtThreshold = buildPayloadOfBytes(topicProperty.getMaxMessageSize()); topicRequests.put(buildRequestEntry(payloadAtThreshold)); await() @@ -645,7 +646,7 @@ void testCanAddPayloadRejectsEntryExceedingSizeThreshold() throws Exception { when(topicProperty.isFifo()).thenReturn(true); context(consumer -> { - final String oversizedPayload = buildPayloadOfBytes(TestableAmazonSnsConsumer.batchSizeBytesThreshold() + 1); + final String oversizedPayload = buildPayloadOfBytes(topicProperty.getMaxMessageSize() + 1); topicRequests.put(buildRequestEntry(oversizedPayload)); await() @@ -662,7 +663,7 @@ void testCanAddPayloadStopsAccumulatingWhenBatchExceedsThreshold() throws Except when(topicProperty.getMaxBatchSize()).thenReturn(10); context(consumer -> { - final int halfThreshold = TestableAmazonSnsConsumer.batchSizeBytesThreshold() / 2; + final int halfThreshold = topicProperty.getMaxMessageSize() / 2; topicRequests.put(buildRequestEntry(buildPayloadOfBytes(halfThreshold))); topicRequests.put(buildRequestEntry(buildPayloadOfBytes(halfThreshold))); topicRequests.put(buildRequestEntry("small-overflow")); @@ -681,7 +682,7 @@ void testCanAddPayloadPublishesFirstEntryAloneWhenItFillsThreshold() throws Exce when(topicProperty.getMaxBatchSize()).thenReturn(10); context(consumer -> { - final int fullThreshold = TestableAmazonSnsConsumer.batchSizeBytesThreshold(); + final int fullThreshold = topicProperty.getMaxMessageSize(); topicRequests.put(buildRequestEntry(buildPayloadOfBytes(fullThreshold))); topicRequests.put(buildRequestEntry("second-entry")); @@ -717,7 +718,7 @@ void testCanAddPayloadSplitsBatchWhenCumulativeSizeExceedsThreshold() throws Exc when(topicProperty.getMaxBatchSize()).thenReturn(10); context(consumer -> { - final int chunkSize = (TestableAmazonSnsConsumer.batchSizeBytesThreshold() / 3) + 1; + final int chunkSize = (topicProperty.getMaxMessageSize() / 3) + 1; topicRequests.put(buildRequestEntry(buildPayloadOfBytes(chunkSize))); topicRequests.put(buildRequestEntry(buildPayloadOfBytes(chunkSize))); topicRequests.put(buildRequestEntry(buildPayloadOfBytes(chunkSize))); @@ -735,7 +736,7 @@ void testCanAddPayloadDoesNotPublishEmptyBatchWhenAllEntriesExceedThreshold() th when(topicProperty.isFifo()).thenReturn(true); context(consumer -> { - final String oversizedPayload = buildPayloadOfBytes(TestableAmazonSnsConsumer.batchSizeBytesThreshold() + 100); + final String oversizedPayload = buildPayloadOfBytes(topicProperty.getMaxMessageSize() + 100); topicRequests.put(buildRequestEntry(oversizedPayload)); await() @@ -755,7 +756,7 @@ void testPoisonRequestEntryRemovesFromPendingRequestsAndFailsListenableFuture() when(topicProperty.isFifo()).thenReturn(true); final String poisonId = "poison-id"; - final String oversizedPayload = buildPayloadOfBytes(TestableAmazonSnsConsumer.batchSizeBytesThreshold() + 100); + final String oversizedPayload = buildPayloadOfBytes(topicProperty.getMaxMessageSize() + 100); final RequestEntry poisonEntry = RequestEntry.builder() .withId(poisonId) .withValue(oversizedPayload) @@ -805,7 +806,7 @@ private void context(final UnaryOperator trackingDecorator, final TryCon static class TestableAmazonSnsConsumer extends AbstractAmazonSnsConsumer implements AutoCloseable { - private static final int BATCH_SIZE_BYTES_THRESHOLD = 256 * 1024; + public static final int KB = 1024; private final AtomicInteger publishCallCount = new AtomicInteger(0); private final AtomicInteger handleErrorCallCount = new AtomicInteger(0); @@ -882,10 +883,6 @@ int getTotalPublishedEntries() { return publishedBatchSizes.stream().mapToInt(Integer::intValue).sum(); } - static int batchSizeBytesThreshold() { - return BATCH_SIZE_BYTES_THRESHOLD; - } - @Override public void close() { shutdown();