From 917bacffc6d817d0b83caafc916e3a150d1c726d Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:46:00 +0100 Subject: [PATCH 1/2] Fix ReactorSender treating absent max-message-size as zero-byte limit Per AMQP 1.0 section 2.7.3, an absent or zero max-message-size on the remote ATTACH frame means the peer imposes no message size limit. The sender previously left linkSize at 0 in that case, so every send failed with "Size of the payload exceeded maximum message size: 0 kb" against brokers that do not advertise the field (e.g. ActiveMQ Artemis). Fall back to ClientConstants.MAX_MESSAGE_LENGTH_BYTES (256 KB) when the field is absent or zero, and clamp advertised values larger than Integer.MAX_VALUE instead of truncating them to negative ints. Fixes #49847 --- sdk/core/azure-core-amqp/CHANGELOG.md | 7 ++ .../amqp/implementation/ReactorSender.java | 18 +++- .../implementation/ReactorSenderTest.java | 90 +++++++++++++++++++ 3 files changed, 111 insertions(+), 4 deletions(-) diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index 990eaf0caef5..b97e0f76f08b 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/sdk/core/azure-core-amqp/CHANGELOG.md @@ -8,6 +8,13 @@ ### 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. + ([#49847](https://github.com/Azure/azure-sdk-for-java/issues/49847)) + ### Other Changes ## 2.12.0 (2026-06-08) diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java index ece618d76ccc..67822884da89 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java +++ b/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,20 @@ public Mono 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); + final long remoteValue = remoteMaxMessageSize.longValue(); + // An unsigned long above Long.MAX_VALUE is negative here; clamp anything above + // Integer.MAX_VALUE so the int-sized encode buffer arithmetic cannot overflow. + linkSize = (remoteValue < 0 || remoteValue > Integer.MAX_VALUE) + ? Integer.MAX_VALUE + : (int) remoteValue; } return linkSize; diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSenderTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSenderTest.java index f05c9c4b9b20..fb9f26ab519a 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSenderTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSenderTest.java @@ -186,6 +186,96 @@ 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); + } + + /** + * 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. From 44ffb208085ae20aaa122bb40ce7ab22e41fabfe Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:58:15 +0100 Subject: [PATCH 2/2] Share max-message-size clamping between getLinkSize and getMaxBatchSize Extract the unsigned-long-to-int saturation into a clampToInt helper and use it in both places that read the remote max-message-size, so a peer advertising a value above Integer.MAX_VALUE cannot truncate the batch size fallback to a negative int either. getMaxBatchSize still returns 0 when the field is absent, preserving its documented contract. Also fixes the vendor-property comment that claimed intValue() matched getLinkSize. --- sdk/core/azure-core-amqp/CHANGELOG.md | 3 ++- .../amqp/implementation/ReactorSender.java | 26 ++++++++++++------- .../implementation/ReactorSenderTest.java | 6 +++++ 3 files changed, 25 insertions(+), 10 deletions(-) diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index b97e0f76f08b..ec58ec66faaa 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/sdk/core/azure-core-amqp/CHANGELOG.md @@ -12,7 +12,8 @@ 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. + 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 diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java index 67822884da89..cd5f7e239f9f 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java @@ -406,12 +406,7 @@ public Mono getLinkSize() { + "Using default link size: {}.", MAX_MESSAGE_LENGTH_BYTES); linkSize = MAX_MESSAGE_LENGTH_BYTES; } else { - final long remoteValue = remoteMaxMessageSize.longValue(); - // An unsigned long above Long.MAX_VALUE is negative here; clamp anything above - // Integer.MAX_VALUE so the int-sized encode buffer arithmetic cannot overflow. - linkSize = (remoteValue < 0 || remoteValue > Integer.MAX_VALUE) - ? Integer.MAX_VALUE - : (int) remoteValue; + linkSize = clampToInt(remoteMaxMessageSize); } return linkSize; @@ -443,8 +438,8 @@ public Mono 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(); } @@ -460,7 +455,7 @@ public Mono getMaxBatchSize() { + "falling back to max-message-size: {}.", AmqpConstants.MAX_MESSAGE_BATCH_SIZE, remoteMaxMessageSize); if (remoteMaxMessageSize != null) { - maxBatchSize = remoteMaxMessageSize.intValue(); + maxBatchSize = clampToInt(remoteMaxMessageSize); } } @@ -469,6 +464,19 @@ public Mono 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(); diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSenderTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSenderTest.java index fb9f26ab519a..25a7c848a9dd 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSenderTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSenderTest.java @@ -251,6 +251,12 @@ public void testLinkSizeClampsRemoteMaxMessageSizeAboveIntegerMax() { .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); } /**