diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index 990eaf0caef5..ec58ec66faaa 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/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) 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..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 @@ -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 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 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 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 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 f05c9c4b9b20..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 @@ -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.