github-actions · Sep 15, 2026 · Sep 15, 2026 · Sep 18, 2026 · Sep 20, 2026 · Sep 20, 2026
@@ -68,11 +68,6 @@ abstract class AbstractAmazonSnsConsumer<C, R, O, E> 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<RequestEntry<E>> request
    * @return true if the request can be added
    */
   private boolean canAddToBatch(final int batchSizeBytes, final int requestEntriesSize, final RequestEntry<E> 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<R> createBatch(final BlockingQueue<RequestEntry<E>> 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)) {

@@ -74,11 +74,14 @@ public ListenableFuture<ResponseSuccessEntry, ResponseFailEntry> 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()
       );
 

@@ -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<V> {
    */
   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));
   }
 
   /**

@@ -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,44 +179,47 @@ static class MessageAttributesInternal extends AbstractMessageAttributes<Integer
      */
     public static final MessageAttributesInternal INSTANCE = new MessageAttributesInternal();
 
+    private final ToIntFunction<Class<? extends Number>> 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;
     }
 
     /**
      * {@inheritDoc}
      */
     @Override
     public Integer getStringMessageAttribute(final String value) {
-      return value.getBytes(StandardCharsets.UTF_8).length;
+      return STRING_SIZE + value.getBytes(StandardCharsets.UTF_8).length;
     }
 
     /**
      * {@inheritDoc}
      */
     @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());
     }
 
     /**
      * {@inheritDoc}
      */
     @Override
     public Integer getBinaryMessageAttribute(final ByteBuffer value) {
-      return value.remaining();
+      return BINARY_SIZE + value.remaining();
     }
 
     /**
      * {@inheritDoc}
      */
     @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;
     }
 
   }

@@ -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.
  * <p>
  * 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

@@ -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<TopicProperty> {
 
@@ -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.
+     *
+     * <p>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);