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 all four batch creation paths in ServiceBusSenderAsyncClient at 1 MB
(MAX_BATCH_SIZE_BYTES). When the link reports a smaller size, use the link
size. 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,11 @@ 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 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 batches are rejected above 1 MB.
+    // 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 +468,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).",
                         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);
@@ -812,7 +817,7 @@ private Mono<Long> scheduleMessageInternal(ServiceBusMessage message, OffsetDate
 
         return tracer.traceScheduleMono("ServiceBus.scheduleMessage",
             getSendLinkWithRetry("schedule-message").flatMap(link -> link.getLinkSize().flatMap(size -> {
-                final int maxSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES;
+                final int maxSize = Math.min(size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES, MAX_BATCH_SIZE_BYTES);
                 return connectionProcessor.flatMap(connection -> connection.getManagementNode(entityName, entityType))
                     .flatMap(
                         managementNode -> managementNode
@@ -885,7 +890,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 +960,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) {