github-actions · Sep 15, 2026 · Sep 15, 2026 · Sep 18, 2026 · Sep 20, 2026 · Sep 20, 2026
@@ -68,11 +68,6 @@ abstract class AbstractAmazonSqsConsumer<C, R, O, E> implements Runnable, Amazon
    */
   private static final Integer KB = 1024;
 
-  /**
-   * Maximum batch size threshold of 1024 KB imposed by Amazon SQS.
-   */
-  private static final Integer BATCH_SIZE_BYTES_THRESHOLD = 1024 * KB;
-
   /** Class logger. */
   private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAmazonSqsConsumer.class);
 
@@ -123,7 +118,7 @@ protected AbstractAmazonSqsConsumer(
     this.queueProperty = Objects.requireNonNull(queueProperty, "queueProperty cannot be null");
     this.amazonSqsClient = Objects.requireNonNull(amazonSqsClient, "amazonSqsClient 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.queueRequests = queueRequests;
     this.publishDecorator = publishDecorator;
@@ -265,7 +260,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 < queueProperty.getMaxMessageSize())
       && (requestEntriesSize < queueProperty.getMaxBatchSize())
       && Objects.nonNull(request);
   }
@@ -277,7 +272,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 <= queueProperty.getMaxMessageSize();
   }
 
   /**
@@ -304,8 +299,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 1024KB (1,048,576 bytes).");
+        if (messageSize > queueProperty.getMaxMessageSize()) {
+          final String message = String.format("The maximum allowed message size exceeding %dKB (%,d bytes).", queueProperty.getMaxMessageSize() / KB, queueProperty.getMaxMessageSize());
+          throw PoisonRequestEntryException.fromMaximumAllowedMessage(message);
         }
 
         if (canAddPayload(batchSizeBytes.get() + messageSize)) {

@@ -73,11 +73,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());
 
       return listenableFutureImpl;

@@ -17,6 +17,7 @@
 package com.amazon.sqs.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 headers into typed message attributes.
    *
@@ -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.sqs.messaging.lib.exception.PoisonRequestEntryException;
@@ -177,44 +178,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.sqs.messaging.lib.exception;
 
 /**
  * Base exception indicating that a request entry could not be published because
- * it is invalid, either exceeding the maximum allowed message size or failing to
- * serialize. Concrete subtypes are created via the {@code from*} factory methods.
+ * it is invalid, either exceeding the maximum allowed message size or failing
+ * to serialize. Concrete subtypes are created via the {@code from*} factory
+ * methods.
  */
 public abstract class PoisonRequestEntryException extends Exception {
 

@@ -51,8 +51,16 @@
 @AllArgsConstructor(access = AccessLevel.PRIVATE)
 public class QueueProperty {
 
+  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 queue is a FIFO queue.
    */
@@ -78,6 +86,11 @@ public class QueueProperty {
    */
   private final int maxBatchSize;
 
+  /**
+   * The maximum message size in bytes.
+   */
+  private final int maxMessageSize;
+
   /**
    * Validates {@link QueueProperty} instances against the SQS constraints for
    * pool size, queue URL format, linger, batch size, and FIFO consistency.
@@ -110,6 +123,10 @@ public void rules() {
         .must(greaterThanOrEqual(DEFAULT_LINGER))
           .withMessage("'linger' must be greater than or equal to 10 (ten)");
 
+      ruleFor("maxMessageSize", QueueProperty::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", QueueProperty::getMaxBatchSize)
         .must(betweenInclusive(1, 10))
           .withMessage("'maxBatchSize' must be in the range of 1 (one) to 10 (ten)");
@@ -148,6 +165,14 @@ public static class QueuePropertyBuilder {
      */
     private boolean linger$set;
 
+    /**
+     * Tracks whether {@code maxMessageSize(int)} was explicitly invoked.
+     *
+     * <p>This flag allows applying {@link QueueProperty#DEFAULT_MESSAGE_SIZE} only when
+     * no explicit value was provided through the builder.
+     */
+    private boolean maxMessageSize$set;
+
     /**
      * Sets the batching linger time in milliseconds.
      *
@@ -160,6 +185,18 @@ public QueuePropertyBuilder 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 QueuePropertyBuilder maxMessageSize(final int maxMessageSize) {
+      this.maxMessageSize = maxMessageSize;
+      maxMessageSize$set = true;
+      return this;
+    }
+
     /**
      * Builds the {@link QueueProperty}, applying the default linger value if none
      * was explicitly set and validating the result.
@@ -169,8 +206,9 @@ public QueuePropertyBuilder linger(final long linger) {
      */
     public QueueProperty build() {
       final long linger = linger$set ? this.linger : DEFAULT_LINGER;
+      final int maxMessageSize = maxMessageSize$set ? this.maxMessageSize : DEFAULT_MESSAGE_SIZE;
 
-      final QueueProperty queueProperty = new QueueProperty(fifo, maximumPoolSize, queueUrl, linger, maxBatchSize);
+      final QueueProperty queueProperty = new QueueProperty(fifo, maximumPoolSize, queueUrl, linger, maxBatchSize, maxMessageSize);
 
       final ValidationResult validationResult = QueuePropertyValidator.INSTANCE.validate(queueProperty);