Cap ServiceBusMessageBatch size at 1 MB to match broker enforcement

The Service Bus broker enforces a 1 MB batch size limit regardless of the
max-message-size advertised on the AMQP link. Premium partitioned namespaces
advertise 100 MB on the link, causing tryAddMessage() to accept batches the
broker will reject.

Cap the three batch-sending paths in ServiceBusSenderAsyncClient at 1 MB
(MAX_BATCH_SIZE_BYTES): createMessageBatch, sendFluxInternal, and
AmqpMessageCollector. The single-message scheduleMessageInternal path is
not capped since the 1 MB limit is batch-specific and individual messages
on Premium can validly exceed 1 MB.

When a user requests a batch size exceeding 1 MB via
CreateMessageBatchOptions, throw ServiceBusException.

Tracking: azure-service-bus#708
ICM: 51000000793879

Author: EldertGrootenboerMS

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClient.java

@@ -222,6 +222,14 @@ public final class ServiceBusSenderAsyncClient implements AutoCloseable {
      * The default maximum allowable size, in bytes, for a batch to be sent.
      */
     static final int MAX_MESSAGE_LENGTH_BYTES = 256 * 1024;
+    // Temporary workaround: Service Bus enforces a maximum batch payload size of 1 MB that is not
+    // communicated via the AMQP link's max-message-size property. The link reports the per-message
+    // limit (up to 100 MB for Premium partitioned), but the broker rejects batch sends above 1 MB.
+    // This cap is applied only to batch-sending code paths (createMessageBatch, sendFlux,
+    // AmqpMessageCollector). Single-message paths (e.g. scheduleMessage) use the link-reported
+    // per-message limit, which is valid for individual messages on Premium namespaces.
+    // Tracked by: https://github.com/Azure/azure-service-bus/issues/708
+    static final int MAX_BATCH_SIZE_BYTES = 1024 * 1024;
     private static final String TRANSACTION_LINK_NAME = "coordinator";
     private static final ServiceBusMessage END = new ServiceBusMessage(new byte[0]);
     private static final CreateMessageBatchOptions DEFAULT_BATCH_OPTIONS = new CreateMessageBatchOptions();
@@ -463,15 +471,15 @@ public Mono<ServiceBusMessageBatch> createMessageBatch(CreateMessageBatchOptions
         final int maxSize = options.getMaximumSizeInBytes();
 
         return getSendLinkWithRetry("create-batch").flatMap(link -> link.getLinkSize().flatMap(size -> {
-            final int maximumLinkSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES;
+            final int maximumLinkSize = Math.min(size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES, MAX_BATCH_SIZE_BYTES);
             if (maxSize > maximumLinkSize) {
                 return monoError(logger,
                     new IllegalArgumentException(String.format(Locale.US,
-                        "CreateMessageBatchOptions.getMaximumSizeInBytes (%s bytes) is larger than the link size"
-                            + " (%s bytes).",
+                        "CreateMessageBatchOptions.getMaximumSizeInBytes (%s bytes) is larger than the maximum"
+                            + " allowed size (%s bytes).",
                         maxSize, maximumLinkSize)));
             }
-            final int batchSize = maxSize > 0 ? maxSize : maximumLinkSize;
+            final int batchSize = maxSize > 0 ? Math.min(maxSize, maximumLinkSize) : maximumLinkSize;
             return Mono
                 .just(new ServiceBusMessageBatch(isV2, batchSize, link::getErrorContext, tracer, messageSerializer));
         })).onErrorMap(this::mapError);
@@ -885,7 +893,7 @@ private Mono<Void> sendFluxInternal(Flux<ServiceBusMessage> messages,
 
         final Mono<List<ServiceBusMessageBatch>> batchList
             = getSendLinkWithRetry("send-batches").flatMap(link -> link.getLinkSize().flatMap(size -> {
-                final int batchSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES;
+                final int batchSize = Math.min(size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES, MAX_BATCH_SIZE_BYTES);
                 final CreateMessageBatchOptions batchOptions
                     = new CreateMessageBatchOptions().setMaximumSizeInBytes(batchSize);
                 return messages.collect(
@@ -955,8 +963,9 @@ private static class AmqpMessageCollector
         AmqpMessageCollector(boolean isV2, CreateMessageBatchOptions options, Integer maxNumberOfBatches,
             ErrorContextProvider contextProvider, ServiceBusTracer tracer, MessageSerializer serializer) {
             this.maxNumberOfBatches = maxNumberOfBatches;
-            this.maxMessageSize
-                = options.getMaximumSizeInBytes() > 0 ? options.getMaximumSizeInBytes() : MAX_MESSAGE_LENGTH_BYTES;
+            this.maxMessageSize = Math.min(
+                options.getMaximumSizeInBytes() > 0 ? options.getMaximumSizeInBytes() : MAX_MESSAGE_LENGTH_BYTES,
+                MAX_BATCH_SIZE_BYTES);
             this.contextProvider = contextProvider;
             this.tracer = tracer;
             this.serializer = serializer;

sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientTest.java

@@ -75,6 +75,7 @@
 import static com.azure.core.util.tracing.Tracer.HOST_NAME_KEY;
 import static com.azure.core.util.tracing.Tracer.PARENT_TRACE_CONTEXT_KEY;
 import static com.azure.core.util.tracing.Tracer.SPAN_CONTEXT_KEY;
+import static com.azure.messaging.servicebus.ServiceBusSenderAsyncClient.MAX_BATCH_SIZE_BYTES;
 import static com.azure.messaging.servicebus.ServiceBusSenderAsyncClient.MAX_MESSAGE_LENGTH_BYTES;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -302,6 +303,105 @@ void createsMessageBatchWithSize(boolean isV2) {
         }).expectComplete().verify(DEFAULT_TIMEOUT);
     }
 
+    /**
+     * Verifies that the batch max size is capped at MAX_BATCH_SIZE_BYTES (1 MB) when the link reports a larger size.
+     * This simulates a Premium partitioned namespace where the link advertises up to 100 MB per-message.
+     */
+    @ParameterizedTest
+    @MethodSource("selectStack")
+    void createBatchCappedAtMaxBatchSizeWhenLinkReportsLargerSize(boolean isV2) {
+        // Arrange
+        arrangeIfV2(isV2);
+        int largeLinkSize = 100 * 1024 * 1024; // 100 MB
+
+        final AmqpSendLink link = mock(AmqpSendLink.class);
+        when(link.getLinkSize()).thenReturn(Mono.just(largeLinkSize));
+
+        when(connection.createSendLink(eq(ENTITY_NAME), eq(ENTITY_NAME), any(AmqpRetryOptions.class), isNull(),
+            eq(CLIENT_IDENTIFIER))).thenReturn(Mono.just(link));
+
+        // Act & Assert
+        StepVerifier.create(sender.createMessageBatch()).assertNext(batch -> {
+            Assertions.assertEquals(MAX_BATCH_SIZE_BYTES, batch.getMaxSizeInBytes());
+        }).expectComplete().verify(DEFAULT_TIMEOUT);
+    }
+
+    /**
+     * Verifies that the batch max size uses the link size when it is smaller than MAX_BATCH_SIZE_BYTES (1 MB).
+     * This simulates a Standard namespace where the link advertises 256 KB.
+     */
+    @ParameterizedTest
+    @MethodSource("selectStack")
+    void createBatchUsesLinkSizeWhenSmallerThanMaxBatchSize(boolean isV2) {
+        // Arrange
+        arrangeIfV2(isV2);
+        int smallLinkSize = 256 * 1024; // 256 KB
+
+        final AmqpSendLink link = mock(AmqpSendLink.class);
+        when(link.getLinkSize()).thenReturn(Mono.just(smallLinkSize));
+
+        when(connection.createSendLink(eq(ENTITY_NAME), eq(ENTITY_NAME), any(AmqpRetryOptions.class), isNull(),
+            eq(CLIENT_IDENTIFIER))).thenReturn(Mono.just(link));
+
+        // Act & Assert
+        StepVerifier.create(sender.createMessageBatch()).assertNext(batch -> {
+            Assertions.assertEquals(smallLinkSize, batch.getMaxSizeInBytes());
+        }).expectComplete().verify(DEFAULT_TIMEOUT);
+    }
+
+    /**
+     * Verifies that user-specified maxSize exceeding the effective 1 MB cap throws an error.
+     */
+    @ParameterizedTest
+    @MethodSource("selectStack")
+    void createBatchWithOptionsExceedingMaxBatchSizeCapThrowsError(boolean isV2) {
+        // Arrange
+        arrangeIfV2(isV2);
+        int largeLinkSize = 100 * 1024 * 1024; // 100 MB
+        int requestedBatchSize = 2 * 1024 * 1024; // 2 MB - exceeds 1 MB cap
+
+        final AmqpSendLink link = mock(AmqpSendLink.class);
+        when(link.getLinkSize()).thenReturn(Mono.just(largeLinkSize));
+
+        when(connection.createSendLink(eq(ENTITY_NAME), eq(ENTITY_NAME), any(AmqpRetryOptions.class), isNull(),
+            eq(CLIENT_IDENTIFIER))).thenReturn(Mono.just(link));
+
+        final CreateMessageBatchOptions options
+            = new CreateMessageBatchOptions().setMaximumSizeInBytes(requestedBatchSize);
+
+        // Act & Assert
+        // The IllegalArgumentException from createMessageBatch is wrapped by mapError into ServiceBusException.
+        StepVerifier.create(sender.createMessageBatch(options))
+            .expectError(ServiceBusException.class)
+            .verify(DEFAULT_TIMEOUT);
+    }
+
+    /**
+     * Verifies that user-specified maxSize smaller than the 1 MB cap is respected.
+     */
+    @ParameterizedTest
+    @MethodSource("selectStack")
+    void createBatchWithOptionsSmallerThanMaxBatchSizeCapIsRespected(boolean isV2) {
+        // Arrange
+        arrangeIfV2(isV2);
+        int largeLinkSize = 100 * 1024 * 1024; // 100 MB
+        int requestedBatchSize = 500 * 1024; // 500 KB
+
+        final AmqpSendLink link = mock(AmqpSendLink.class);
+        when(link.getLinkSize()).thenReturn(Mono.just(largeLinkSize));
+
+        when(connection.createSendLink(eq(ENTITY_NAME), eq(ENTITY_NAME), any(AmqpRetryOptions.class), isNull(),
+            eq(CLIENT_IDENTIFIER))).thenReturn(Mono.just(link));
+
+        final CreateMessageBatchOptions options
+            = new CreateMessageBatchOptions().setMaximumSizeInBytes(requestedBatchSize);
+
+        // Act & Assert
+        StepVerifier.create(sender.createMessageBatch(options)).assertNext(batch -> {
+            Assertions.assertEquals(requestedBatchSize, batch.getMaxSizeInBytes());
+        }).expectComplete().verify(DEFAULT_TIMEOUT);
+    }
+
     @ParameterizedTest
     @MethodSource("selectStack")
     void scheduleMessageSizeTooBig(boolean isV2) {