Fix ReactorSender treating absent max-message-size on ATTACH as a zero-byte limit

sdk/core/azure-core-amqp/CHANGELOG.md

@@ -8,6 +8,14 @@
 
 ### Bugs Fixed
 
+- Fixed `ReactorSender` treating an absent `max-message-size` on the remote ATTACH frame as a zero-byte limit,
+  which failed every send with "Size of the payload exceeded maximum message size: 0 kb" against brokers that
+  do not advertise the field (for example ActiveMQ Artemis). Per AMQP 1.0 section 2.7.3, an absent or zero
+  value means the peer imposes no limit; the link now falls back to a 256 KB default in that case. Remote
+  values larger than `Integer.MAX_VALUE` are now clamped instead of overflowing the link size and the
+  max-message-size fallback of the batch size.
+  ([#49847](https://github.com/Azure/azure-sdk-for-java/issues/49847))
+
 ### Other Changes
 
 ## 2.12.0 (2026-06-08)

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java

@@ -67,6 +67,7 @@
 import static com.azure.core.amqp.implementation.ClientConstants.ENTITY_PATH_KEY;
 import static com.azure.core.amqp.implementation.ClientConstants.LINK_NAME_KEY;
 import static com.azure.core.amqp.implementation.ClientConstants.MAX_AMQP_HEADER_SIZE_BYTES;
+import static com.azure.core.amqp.implementation.ClientConstants.MAX_MESSAGE_LENGTH_BYTES;
 import static com.azure.core.amqp.implementation.ClientConstants.NOT_APPLICABLE;
 import static com.azure.core.amqp.implementation.ClientConstants.SERVER_BUSY_BASE_SLEEP_TIME_IN_SECS;
 import static com.azure.core.util.FluxUtil.monoError;
@@ -397,11 +398,15 @@ public Mono<Integer> getLinkSize() {
                     activeTimeoutMessage)
                 .then(Mono.fromCallable(() -> {
                     final UnsignedLong remoteMaxMessageSize = sender.getRemoteMaxMessageSize();
-                    if (remoteMaxMessageSize != null) {
-                        linkSize = remoteMaxMessageSize.intValue();
+                    if (remoteMaxMessageSize == null || remoteMaxMessageSize.longValue() == 0) {
+                        // Per AMQP 1.0 section 2.7.3, an absent or zero max-message-size on the remote ATTACH
+                        // means the peer imposes no message size limit. Fall back to a bounded client-side
+                        // default rather than treating it as a zero-byte limit that fails every send.
+                        logger.info("Remote peer did not advertise a max-message-size on link attach. "
+                            + "Using default link size: {}.", MAX_MESSAGE_LENGTH_BYTES);
+                        linkSize = MAX_MESSAGE_LENGTH_BYTES;
                     } else {
-                        logger.warning("Could not get the getRemoteMaxMessageSize. Returning current link size: {}",
-                            linkSize);
+                        linkSize = clampToInt(remoteMaxMessageSize);
                     }
 
                     return linkSize;
@@ -433,8 +438,8 @@ public Mono<Integer> getMaxBatchSize() {
                         && remoteProperties.containsKey(AmqpConstants.MAX_MESSAGE_BATCH_SIZE)) {
                         final Object value = remoteProperties.get(AmqpConstants.MAX_MESSAGE_BATCH_SIZE);
                         // The AMQP property may arrive as UnsignedLong, UnsignedInteger, Long, or Integer.
-                        // intValue() is consistent with getLinkSize() — values > Integer.MAX_VALUE (impossible
-                        // for batch sizes) would overflow to negative and trigger the fallback below.
+                        // Values > Integer.MAX_VALUE (impossible for batch sizes) overflow intValue() to
+                        // negative and trigger the fallback below.
                         if (value instanceof Number) {
                             maxBatchSize = ((Number) value).intValue();
                         }
@@ -450,7 +455,7 @@ public Mono<Integer> getMaxBatchSize() {
                                 + "falling back to max-message-size: {}.",
                             AmqpConstants.MAX_MESSAGE_BATCH_SIZE, remoteMaxMessageSize);
                         if (remoteMaxMessageSize != null) {
-                            maxBatchSize = remoteMaxMessageSize.intValue();
+                            maxBatchSize = clampToInt(remoteMaxMessageSize);
                         }
                     }
 
@@ -459,6 +464,19 @@ public Mono<Integer> getMaxBatchSize() {
         }
     }
 
+    /**
+     * Converts the remote peer's max-message-size to an int, saturating at {@link Integer#MAX_VALUE}. An unsigned
+     * long above {@link Long#MAX_VALUE} reads as a negative long; peers may advertise such values to signal an
+     * effectively unlimited link.
+     *
+     * @param remoteMaxMessageSize the remote max-message-size, not null.
+     * @return the value as an int, or {@link Integer#MAX_VALUE} if it does not fit in an int.
+     */
+    private static int clampToInt(UnsignedLong remoteMaxMessageSize) {
+        final long value = remoteMaxMessageSize.longValue();
+        return (value < 0 || value > Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int) value;
+    }
+
     @Override
     public boolean isDisposed() {
         return isDisposed.get();

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSenderTest.java

@@ -186,6 +186,102 @@ public void testLinkSize() {
         verify(sender).getRemoteMaxMessageSize();
     }
 
+    /**
+     * Verifies that getLinkSize falls back to a bounded default when the remote peer does not advertise
+     * max-message-size on its ATTACH frame. Per AMQP 1.0 section 2.7.3 an absent value means the peer imposes no
+     * limit, so it must not be treated as a zero-byte limit that fails every send.
+     */
+    @Test
+    public void testLinkSizeUsesDefaultWhenRemoteMaxMessageSizeAbsent() {
+        // Arrange -- brokers such as ActiveMQ Artemis omit max-message-size on their receiving links.
+        when(sender.getRemoteMaxMessageSize()).thenReturn(null);
+
+        reactorSender = new ReactorSender(amqpConnection, ENTITY_PATH, sender, handler, reactorProvider, tokenManager,
+            messageSerializer, options, scheduler, AmqpMetricsProvider.noop());
+
+        // Act & Assert
+        StepVerifier.create(reactorSender.getLinkSize())
+            .expectNext(ClientConstants.MAX_MESSAGE_LENGTH_BYTES)
+            .expectComplete()
+            .verify(VERIFY_TIMEOUT);
+
+        // Second call returns the cached default without consulting the link again.
+        StepVerifier.create(reactorSender.getLinkSize())
+            .expectNext(ClientConstants.MAX_MESSAGE_LENGTH_BYTES)
+            .expectComplete()
+            .verify(VERIFY_TIMEOUT);
+
+        verify(sender, times(1)).getRemoteMaxMessageSize();
+    }
+
+    /**
+     * Verifies that getLinkSize treats an explicit zero max-message-size as "no limit imposed" per AMQP 1.0
+     * section 2.7.3 and falls back to the bounded default.
+     */
+    @Test
+    public void testLinkSizeUsesDefaultWhenRemoteMaxMessageSizeZero() {
+        // Arrange
+        when(sender.getRemoteMaxMessageSize()).thenReturn(UnsignedLong.valueOf(0));
+
+        reactorSender = new ReactorSender(amqpConnection, ENTITY_PATH, sender, handler, reactorProvider, tokenManager,
+            messageSerializer, options, scheduler, AmqpMetricsProvider.noop());
+
+        // Act & Assert
+        StepVerifier.create(reactorSender.getLinkSize())
+            .expectNext(ClientConstants.MAX_MESSAGE_LENGTH_BYTES)
+            .expectComplete()
+            .verify(VERIFY_TIMEOUT);
+    }
+
+    /**
+     * Verifies that getLinkSize clamps a remote max-message-size larger than Integer.MAX_VALUE instead of
+     * truncating it to a negative or otherwise incorrect int. Peers may advertise the maximum unsigned long to
+     * signal an effectively unlimited link.
+     */
+    @Test
+    public void testLinkSizeClampsRemoteMaxMessageSizeAboveIntegerMax() {
+        // Arrange -- 0xFFFFFFFFFFFFFFFF is 2^64 - 1, the largest unsigned long.
+        when(sender.getRemoteMaxMessageSize()).thenReturn(new UnsignedLong(0xFFFFFFFFFFFFFFFFL));
+
+        reactorSender = new ReactorSender(amqpConnection, ENTITY_PATH, sender, handler, reactorProvider, tokenManager,
+            messageSerializer, options, scheduler, AmqpMetricsProvider.noop());
+
+        // Act & Assert
+        StepVerifier.create(reactorSender.getLinkSize())
+            .expectNext(Integer.MAX_VALUE)
+            .expectComplete()
+            .verify(VERIFY_TIMEOUT);
+
+        // The max-message-size fallback in getMaxBatchSize applies the same clamping.
+        StepVerifier.create(reactorSender.getMaxBatchSize())
+            .expectNext(Integer.MAX_VALUE)
+            .expectComplete()
+            .verify(VERIFY_TIMEOUT);
+    }
+
+    /**
+     * Verifies that a message is sent successfully when the remote peer does not advertise max-message-size:
+     * the encode buffer is sized from the default link size instead of zero bytes.
+     */
+    @Test
+    public void testSendWhenRemoteMaxMessageSizeAbsent() {
+        // Arrange
+        when(sender.getRemoteMaxMessageSize()).thenReturn(null);
+
+        reactorSender = new ReactorSender(amqpConnection, ENTITY_PATH, sender, handler, reactorProvider, tokenManager,
+            messageSerializer, options, scheduler, AmqpMetricsProvider.noop());
+        final ReactorSender spyReactorSender = spy(reactorSender);
+
+        doReturn(Mono.empty()).when(spyReactorSender).send(any(byte[].class), anyInt(), anyInt(), isNull());
+
+        // Act
+        StepVerifier.create(spyReactorSender.send(message)).expectComplete().verify(VERIFY_TIMEOUT);
+
+        // Assert
+        verify(sender).getRemoteMaxMessageSize();
+        verify(spyReactorSender).send(any(byte[].class), anyInt(), anyInt(), isNull());
+    }
+
     /**
      * Verifies that getMaxBatchSize reads the vendor property com.microsoft:max-message-batch-size from the link's
      * remote properties and returns its value.