diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AmqpChannelProcessor.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AmqpChannelProcessor.java index 3c1eef933a389..764b998b02507 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AmqpChannelProcessor.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AmqpChannelProcessor.java @@ -140,15 +140,19 @@ public void onNext(T amqpChannel) { logger.info("Channel is now active."); } }, error -> { - setAndClearChannel(); - onError(error); + if (setAndClearChannel(amqpChannel)) { + onError(error); + } else { + logger.verbose("Ignoring error from a replaced channel.", error); + } }, () -> { if (isDisposed()) { logger.info("Channel is disposed."); } else { - logger.info("Channel is closed. Requesting upstream."); - setAndClearChannel(); - requestUpstream(); + if (setAndClearChannel(amqpChannel)) { + logger.info("Channel is closed. Requesting upstream."); + requestUpstream(); + } } }); } @@ -347,6 +351,36 @@ private void setAndClearChannel() { close(oldChannel); } + /** + * Force-closes the current cached channel so that the next subscriber receives a fresh one. + * This is used for connection-level recovery when the current connection is stale + * but the processor has not detected it (e.g., heartbeats echoed by intermediate infrastructure). + */ + public void forceCloseChannel() { + setAndClearChannel(); + } + + /** + * Closes the failed channel only if it is still the current channel. + * + * @param failedChannel The channel used by the failed operation. + */ + public void forceCloseChannel(T failedChannel) { + Objects.requireNonNull(failedChannel, "'failedChannel' cannot be null."); + setAndClearChannel(failedChannel); + } + + private boolean setAndClearChannel(T failedChannel) { + synchronized (lock) { + if (currentChannel != failedChannel) { + return false; + } + currentChannel = null; + } + close(failedChannel); + return true; + } + /** * Checks the current state of the channel for this channel and returns true if the channel is null or if this * processor is disposed. diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java index 58ca458207d40..07949665dff08 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java @@ -362,7 +362,7 @@ public Mono createSession(String sessionName) { logger.atInfo() .addKeyValue(SESSION_NAME_KEY, sessionName) .log("Error occurred. Removing and disposing session", error); - removeSession(key); + removeSession(key, amqpSession); }, () -> { // If we were already disposing of the connection, the session would be removed. if (isDisposed.get()) { @@ -372,7 +372,7 @@ public Mono createSession(String sessionName) { logger.atVerbose() .addKeyValue(SESSION_NAME_KEY, sessionName) .log("Complete. Removing and disposing session."); - removeSession(key); + removeSession(key, amqpSession); }); return new SessionSubscription(amqpSession, subscription); @@ -396,8 +396,7 @@ public Mono createSession(String sessionName) { final AmqpException amqpException = (AmqpException) error; if (amqpException.getErrorCondition() == TIMEOUT_ERROR) { - final SessionSubscription removed = sessionMap.remove(sessionName); - removed.dispose(); + removeSession(sessionName, sessionSubscription.getSession()); } }); @@ -443,6 +442,28 @@ public boolean removeSession(String sessionName) { return removed != null; } + /** + * Removes a session only if it is still the cached instance. + * + * @param sessionName The name used to cache the session. + * @param expectedSession The session owned by the failed operation. + * @return Whether the expected session was removed. + */ + protected boolean removeSession(String sessionName, AmqpSession expectedSession) { + if (sessionName == null || expectedSession == null) { + return false; + } + if (useSessionChannelCache) { + return sessionCache.evict(sessionName, expectedSession); + } + final SessionSubscription entry = sessionMap.get(sessionName); + if (entry == null || entry.getSession() != expectedSession || !sessionMap.remove(sessionName, entry)) { + return false; + } + entry.dispose(); + return true; + } + @Override public boolean isDisposed() { return isDisposed.get(); diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnectionCache.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnectionCache.java index d772cfdf06988..b9f2be1215262 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnectionCache.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnectionCache.java @@ -48,6 +48,10 @@ public final class ReactorConnectionCache implement // any dependent type; instead, the dependent type must acquire Connection only through the cache route, // i.e., by subscribing to 'createOrGetCachedConnection' via 'get()' getter. private volatile T currentConnection; + // Holds the ID of the connection that invalidateConnection() asked to force-invalidate. + // Only the connection whose getId() matches this value will be invalidated by cacheInvalidateIf; + // a freshly created connection with a different ID is never accidentally invalidated. + private final AtomicReference forceInvalidateConnectionId = new AtomicReference<>(null); private final State state = new State(); /** @@ -113,12 +117,25 @@ public ReactorConnectionCache(Supplier connectionSupplier, String fullyQualif } }).cacheInvalidateIf(c -> { if (c.isDisposed()) { + // Connection disposed for any reason. Clean up the force-invalidate marker if it + // was targeting this connection so it is not accidentally consumed by a future + // connection that happens to have the same ID. + forceInvalidateConnectionId.compareAndSet(c.getId(), null); withConnectionId(logger, c.getId()).log("The connection is closed, requesting a new connection."); return true; - } else { - // Emit cached connection. - return false; } + final String targetId = forceInvalidateConnectionId.get(); + if (targetId != null + && targetId.equals(c.getId()) + && forceInvalidateConnectionId.compareAndSet(targetId, null)) { + // invalidateConnection() marked this connection dirty. + // Close it here — the cache owns the connection lifecycle. + withConnectionId(logger, c.getId()).log("Force-invalidating and closing connection for recovery."); + closeConnection(c, logger, "Force-invalidated for recovery."); + return true; + } + // No forced invalidation targeted this connection — emit it from cache. + return false; }); } @@ -172,6 +189,49 @@ public boolean isCurrentConnectionClosed() { return (currentConnection != null && currentConnection.isDisposed()) || terminated; } + /** + * Marks the current cached connection for invalidation so that the next {@link #get()} call + * closes it and creates a fresh connection. This is used for connection-level recovery when + * the current connection is in a stale state that the cache's normal error detection (via + * endpoint state signals) has not detected — for example, when intermediate infrastructure + * (load balancers, NAT gateways) is echoing AMQP heartbeats on behalf of a dead connection. + * + *

The actual close is deferred to the cache chain's {@code cacheInvalidateIf} predicate + * inside {@link #get()}, keeping the connection lifecycle local to the cache. This avoids + * the concurrency concern of two threads touching the cached connection simultaneously.

+ * + *

This is modeled after the Go SDK's {@code Namespace.Recover()} which explicitly closes + * the old connection and increments the connection revision.

+ * + *

This method is safe to call concurrently. If the connection is already closed or being + * closed, this is a no-op.

+ */ + public void invalidateConnection() { + final T connection = currentConnection; + if (connection != null && !connection.isDisposed()) { + withConnectionId(logger, connection.getId()) + .log("Marking connection for force-invalidation. Next get() will close and replace it."); + forceInvalidateConnectionId.set(connection.getId()); + } + } + + /** + * Marks the failed connection for invalidation only if it is still cached. + * A delayed failure must not invalidate a healthy replacement connection. + * + * @param connectionId The identifier of the connection used by the failed operation. + */ + public void invalidateConnection(String connectionId) { + Objects.requireNonNull(connectionId, "'connectionId' cannot be null."); + synchronized (lock) { + final T connection = currentConnection; + if (connection != null && connectionId.equals(connection.getId()) && !connection.isDisposed()) { + withConnectionId(logger, connectionId).log("Marking failed connection for force-invalidation."); + forceInvalidateConnectionId.set(connectionId); + } + } + } + /** * Terminate so that consumers will no longer be able to request connection. If there is a current (cached) * connection then it will be closed. diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSessionCache.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSessionCache.java index dcfbaad9bfb59..008bf10cd9396 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSessionCache.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSessionCache.java @@ -3,6 +3,7 @@ package com.azure.core.amqp.implementation; +import com.azure.core.amqp.AmqpSession; import com.azure.core.util.logging.ClientLogger; import org.apache.qpid.proton.engine.Connection; import reactor.core.Disposable; @@ -119,6 +120,18 @@ boolean evict(String name) { return removed != null; } + boolean evict(String name, AmqpSession expectedSession) { + if (name == null || expectedSession == null) { + return false; + } + final Entry entry = entries.get(name); + if (entry == null || entry.getSession() != expectedSession || !entries.remove(name, entry)) { + return false; + } + entry.dispose(); + return true; + } + /** * Signal that the owner ({@link ReactorConnection}) of the cache is disposed of. */ @@ -193,7 +206,7 @@ private void evict(ReactorSession session, String message, Throwable error) { } else { logger.atInfo().addKeyValue(SESSION_NAME_KEY, name).addKeyValue(SESSION_ID_KEY, id).log(message); } - evict(name); + evict(name, session); } /** diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RecoveryKind.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RecoveryKind.java new file mode 100644 index 0000000000000..f7c6c9b9a6a1e --- /dev/null +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RecoveryKind.java @@ -0,0 +1,173 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.amqp.implementation; + +import com.azure.core.amqp.exception.AmqpErrorCondition; +import com.azure.core.amqp.exception.AmqpException; +import com.azure.core.amqp.exception.OperationCancelledException; + +import java.util.Locale; +import java.util.concurrent.TimeoutException; + +/** + * Classifies errors into recovery tiers, determining what resources should be closed + * between retry attempts. This follows the tiered recovery pattern used by the Go, .NET, + * Python, and JS Azure SDKs. + * + *
    + *
  • {@link #NONE} — Retry on the same link (server-busy, timeouts).
    • + *
  • {@link #LINK} — Close the send/receive link; next retry creates a fresh link on the same connection.
    • + *
  • {@link #CONNECTION} — Close the entire connection; next retry creates a fresh connection and link.
    • + *
  • {@link #FATAL} — Do not retry (unauthorized, not-found, message too large).
    • + *
+ */ +public enum RecoveryKind { + /** + * No recovery needed — retry on the same link and connection. + * Applies to: server-busy, timeouts, resource-limit-exceeded. + */ + NONE, + + /** + * Close the link (and its session) before retrying. The next retry creates a fresh link + * on the same connection. + * Applies to: link:detach-forced, link:stolen, transient AMQP errors on the link. + */ + LINK, + + /** + * Close the entire connection before retrying. The next retry creates a fresh connection, + * session, and link. + * Applies to: connection:forced, connection:framing-error, proton:io, internal-error. + */ + CONNECTION, + + /** + * Do not retry — the error is permanent. + * Applies to: unauthorized-access, not-found, message-size-exceeded. + */ + FATAL; + + /** + * Classifies the given error into a {@link RecoveryKind} that determines what resources + * should be invalidated between retry attempts. + * + * @param error The error to classify. + * @return The recovery kind for the given error. + */ + public static RecoveryKind classify(Throwable error) { + if (error == null) { + return NONE; + } + + // Timeouts — retry on same link, the link may still be healthy. + if (error instanceof TimeoutException) { + return NONE; + } + + if (error instanceof OperationCancelledException) { + return LINK; + } + + if (error instanceof AmqpException) { + final AmqpException amqpError = (AmqpException) error; + final AmqpErrorCondition condition = amqpError.getErrorCondition(); + + if (condition != null) { + switch (condition) { + // Connection-level errors — close the entire connection. + case CONNECTION_FORCED: + case CONNECTION_FRAMING_ERROR: + case CONNECTION_REDIRECT: + case PROTON_IO: + case INTERNAL_ERROR: + return CONNECTION; + + // Link-level errors — close the link, keep the connection. + case LINK_DETACH_FORCED: + case LINK_STOLEN: + case LINK_REDIRECT: + case PARTITION_NOT_OWNED_ERROR: + case TRANSFER_LIMIT_EXCEEDED: + // operation-cancelled can signal "AMQP layer unexpectedly aborted or disconnected" + // (e.g. ReceiverUnsettledDeliveries remote Released outcome), requiring link recovery. + case OPERATION_CANCELLED: + return LINK; + + // Fatal errors — do not retry. + case NOT_FOUND: + case UNAUTHORIZED_ACCESS: + case LINK_PAYLOAD_SIZE_EXCEEDED: + case NOT_ALLOWED: + case NOT_IMPLEMENTED: + case ENTITY_DISABLED_ERROR: + case ENTITY_ALREADY_EXISTS: + case PUBLISHER_REVOKED_ERROR: + case ARGUMENT_ERROR: + case ARGUMENT_OUT_OF_RANGE_ERROR: + case ILLEGAL_STATE: + case MESSAGE_LOCK_LOST: + case STORE_LOCK_LOST_ERROR: + case SESSION_CANNOT_BE_LOCKED: + return FATAL; + + // Server-busy, timeouts, and resource-limit errors — retry on same link. + // RESOURCE_LIMIT_EXCEEDED is treated as transient here because ReactorSender + // groups it alongside SERVER_BUSY and TIMEOUT in its send-error retry logic. + case SERVER_BUSY_ERROR: + case TIMEOUT_ERROR: + case RESOURCE_LIMIT_EXCEEDED: + return NONE; + + // Session/lock errors — link-level recovery. + // Session lock loss means the session link is invalid and + // a fresh link must be acquired for a new session. + case SESSION_LOCK_LOST: + case SESSION_NOT_FOUND: + case MESSAGE_NOT_FOUND: + return LINK; + + default: + break; + } + } + + // Transient AMQP errors without a specific condition — link recovery. + if (amqpError.isTransient()) { + return LINK; + } + + // Non-transient AMQP errors without a recognized condition — fatal. + return FATAL; + } + + // RequestResponseChannelClosedException — link-level (parent connection disposing). + if (error instanceof RequestResponseChannelClosedException) { + return LINK; + } + + // IllegalStateException thrown by a disposed ReactorSender (e.g., "Cannot publish + // message when disposed." or "Cannot publish data batch when disposed."). This is + // a link-staleness signal: the link was closed (possibly by a concurrent recovery + // path) before the in-flight send could complete. LINK recovery creates a fresh + // link on the next retry. + // Match both "Cannot publish" and "disposed" to avoid misclassifying unrelated + // disposal signals (e.g., "Connection is disposed. Cannot get management instance."). + if (error instanceof IllegalStateException) { + final String msg = error.getMessage(); + if (msg != null) { + final String normalizedMsg = msg.toLowerCase(Locale.ROOT); + if (normalizedMsg.contains("cannot publish") && normalizedMsg.contains("disposed")) { + return LINK; + } + } + } + + // Unknown non-AMQP errors — treat as fatal (don't retry application or SDK bugs). + // The Go SDK defaults to CONNECTION for unknown errors, but those are AMQP-layer + // errors (io.EOF, net.Error). Java's non-AMQP exceptions (e.g., AzureException, + // RuntimeException) should fail fast rather than trigger connection recovery. + return FATAL; + } +} diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RetryUtil.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RetryUtil.java index c22c8ac928b3b..6b6c5bd57cf6a 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RetryUtil.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RetryUtil.java @@ -129,4 +129,5 @@ static Retry createRetry(AmqpRetryOptions options) { .filter(error -> error instanceof TimeoutException || (error instanceof AmqpException && ((AmqpException) error).isTransient())); } + } diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/AmqpChannelProcessorTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/AmqpChannelProcessorTest.java index 6f6a351ab7766..e81c75c9d90bc 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/AmqpChannelProcessorTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/AmqpChannelProcessorTest.java @@ -10,12 +10,15 @@ import com.azure.core.amqp.exception.AmqpErrorCondition; import com.azure.core.amqp.exception.AmqpErrorContext; import com.azure.core.amqp.exception.AmqpException; +import com.azure.core.util.AsyncCloseable; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; import org.reactivestreams.Subscription; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; import reactor.test.StepVerifier; import reactor.test.publisher.TestPublisher; @@ -30,7 +33,12 @@ import java.util.stream.Stream; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; /** * Tests for {@link AmqpChannelProcessor}. @@ -353,10 +361,48 @@ private static AmqpChannelProcessor createChannelProcessor() { return createChannelProcessor(new FixedAmqpRetryPolicy(new AmqpRetryOptions())); } + @Test + void targetedClosePreservesReplacementChannel() { + final AmqpChannelProcessor processor = createChannelProcessor(); + try { + processor.onNext(connection1); + processor.onNext(connection2); + processor.forceCloseChannel(connection1); + StepVerifier.create(processor).expectNext(connection2).verifyComplete(); + processor.forceCloseChannel(connection2); + assertTrue(processor.isChannelClosed()); + } finally { + processor.dispose(); + } + } + private static AmqpChannelProcessor createChannelProcessor(AmqpRetryPolicy retryPolicy) { return new AmqpChannelProcessor<>("namespace-test", TestObject::getStates, retryPolicy, new HashMap<>()); } + @Test + void closingOldChannelMustNotClearReplacement() { + final AsyncCloseable first = mock(AsyncCloseable.class); + final AsyncCloseable replacement = mock(AsyncCloseable.class); + final Sinks.Many states = Sinks.many().replay().latestOrDefault(AmqpEndpointState.ACTIVE); + when(first.closeAsync()).thenAnswer(invocation -> { + states.tryEmitComplete(); + return Mono.empty(); + }); + when(replacement.closeAsync()).thenReturn(Mono.empty()); + final AmqpChannelProcessor processor + = new AmqpChannelProcessor<>("namespace-test", channel -> channel == first ? states.asFlux() : Flux.never(), + new FixedAmqpRetryPolicy(new AmqpRetryOptions()), new HashMap<>()); + try { + processor.onNext(first); + processor.onNext(replacement); + assertFalse(processor.isChannelClosed()); + verify(replacement, never()).closeAsync(); + } finally { + processor.dispose(); + } + } + static final class TestObject { private final TestPublisher processor = TestPublisher.createCold(); diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionCacheTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionCacheTest.java index 4b51308f8e569..e6f18d2e81146 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionCacheTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionCacheTest.java @@ -222,6 +222,132 @@ public void shouldRefreshCacheOnErrorInCachedConnection() { } } + @Test + public void invalidateConnectionReplacesCachedConnectionOnNextGet() { + final ConnectionSupplier connectionSupplier = new ConnectionSupplier(); + final ReactorConnectionCache connectionCache + = new ReactorConnectionCache<>(connectionSupplier, FQDN, ENTITY_PATH, retryPolicy, new HashMap<>()); + try { + final ReactorConnection[] c = new ReactorConnection[1]; + final Mono connectionMono = connectionCache.get(); + // Populate the cache with the first connection. + StepVerifier.create(connectionMono, 0) + .thenRequest(1) + .then(() -> connectionSupplier.emitEndpointState(EndpointState.ACTIVE)) + .expectNextMatches(con -> { + Assertions.assertFalse(con.isDisposed()); + c[0] = con; + return true; + }) + .expectComplete() + .verify(VERIFY_TIMEOUT); + connectionSupplier.assertInvocationCount(1); + + // Mark the connection used by the failed operation. + connectionCache.invalidateConnection(c[0].getId()); + + // Next subscription must close the marked connection and serve a brand-new one. + StepVerifier.create(connectionMono, 0) + .thenRequest(1) + .then(() -> connectionSupplier.emitEndpointState(EndpointState.ACTIVE)) + .expectNextMatches(con -> { + Assertions.assertFalse(con.isDisposed()); + Assertions.assertNotEquals(c[0], con); + return true; + }) + .expectComplete() + .verify(VERIFY_TIMEOUT); + + // The previously cached connection should be closed by the cacheInvalidateIf branch. + Assertions.assertTrue(c[0].isDisposed()); + // A second connection had to be supplied because the first was force-invalidated. + connectionSupplier.assertInvocationCount(2); + } finally { + connectionSupplier.dispose(); + connectionCache.dispose(); + } + } + + @Test + public void invalidateConnectionDoesNotInvalidateNewlyCreatedConnection() { + // Asserts the AB-A safety property: invalidate() targets a specific connection-id, so a + // fresh connection created after the invalidation marker was consumed must NOT be evicted + // by a stale marker. + final ConnectionSupplier connectionSupplier = new ConnectionSupplier(); + final ReactorConnectionCache connectionCache + = new ReactorConnectionCache<>(connectionSupplier, FQDN, ENTITY_PATH, retryPolicy, new HashMap<>()); + try { + final ReactorConnection[] c = new ReactorConnection[2]; + final Mono connectionMono = connectionCache.get(); + StepVerifier.create(connectionMono, 0) + .thenRequest(1) + .then(() -> connectionSupplier.emitEndpointState(EndpointState.ACTIVE)) + .expectNextMatches(con -> { + c[0] = con; + return true; + }) + .expectComplete() + .verify(VERIFY_TIMEOUT); + + connectionCache.invalidateConnection(); + + // Consume the invalidation marker and obtain the replacement connection. + StepVerifier.create(connectionMono, 0) + .thenRequest(1) + .then(() -> connectionSupplier.emitEndpointState(EndpointState.ACTIVE)) + .expectNextMatches(con -> { + c[1] = con; + Assertions.assertNotEquals(c[0], c[1]); + return true; + }) + .expectComplete() + .verify(VERIFY_TIMEOUT); + + // A delayed error from the old connection must not evict its replacement. + connectionCache.invalidateConnection(c[0].getId()); + + // The marker has been consumed; another get() must serve the same (cached) replacement + // connection without creating a third one. + StepVerifier.create(connectionMono, 0).thenRequest(1).expectNextMatches(con -> { + Assertions.assertEquals(c[1], con); + Assertions.assertFalse(con.isDisposed()); + return true; + }).expectComplete().verify(VERIFY_TIMEOUT); + + connectionSupplier.assertInvocationCount(2); + } finally { + connectionSupplier.dispose(); + connectionCache.dispose(); + } + } + + @Test + public void invalidateConnectionWithoutCachedConnectionIsNoOp() { + final ConnectionSupplier connectionSupplier = new ConnectionSupplier(); + final ReactorConnectionCache connectionCache + = new ReactorConnectionCache<>(connectionSupplier, FQDN, ENTITY_PATH, retryPolicy, new HashMap<>()); + try { + // No connection has been supplied yet. + connectionCache.invalidateConnection(); + connectionSupplier.assertInvocationCount(0); + + // First get() after the no-op invalidate() must still produce a connection normally. + StepVerifier.create(connectionCache.get(), 0) + .thenRequest(1) + .then(() -> connectionSupplier.emitEndpointState(EndpointState.ACTIVE)) + .expectNextMatches(con -> { + Assertions.assertFalse(con.isDisposed()); + return true; + }) + .expectComplete() + .verify(VERIFY_TIMEOUT); + connectionSupplier.assertInvocationCount(1); + } finally { + connectionSupplier.dispose(); + connectionCache.dispose(); + } + } + @Test public void shouldBubbleUpNonRetriableError() { final ConnectionSupplier connectionSupplier = new ConnectionSupplier(); diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java index 20c36c9f0d71c..610d7bf6ea4a1 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java @@ -6,6 +6,7 @@ import com.azure.core.amqp.AmqpEndpointState; import com.azure.core.amqp.AmqpRetryMode; import com.azure.core.amqp.AmqpRetryOptions; +import com.azure.core.amqp.AmqpSession; import com.azure.core.amqp.AmqpTransportType; import com.azure.core.amqp.ProxyOptions; import com.azure.core.amqp.exception.AmqpErrorCondition; @@ -409,6 +410,35 @@ void removeSessionThatDoesNotExist() { verify(record).set(Handler.class, Handler.class, sessionHandler); } + @Test + void removeSessionOnlyRemovesExpectedInstance() { + final Session secondSession = mock(Session.class); + final Event secondEvent = mock(Event.class); + when(connectionProtonJ.session()).thenReturn(session).thenReturn(secondSession); + when(session.attachments()).thenReturn(record); + when(secondSession.attachments()).thenReturn(mock(Record.class)); + when(session.getRemoteState()).thenReturn(EndpointState.ACTIVE); + when(secondSession.getRemoteState()).thenReturn(EndpointState.ACTIVE); + when(secondEvent.getSession()).thenReturn(secondSession); + when(connectionProtonJ.getRemoteState()).thenReturn(EndpointState.ACTIVE); + connectionHandler.onConnectionRemoteOpen(connectionEvent); + sessionHandler.onSessionRemoteOpen(sessionEvent); + final AmqpSession first = connection.createSession(SESSION_NAME).block(VERIFY_TIMEOUT); + assertNotNull(first); + assertFalse(connection.removeSession(SESSION_NAME, mock(AmqpSession.class))); + assertSame(first, connection.createSession(SESSION_NAME).block(VERIFY_TIMEOUT)); + assertTrue(connection.removeSession(SESSION_NAME, first)); + sessionHandler2.onSessionRemoteOpen(secondEvent); + final AmqpSession replacement = connection.createSession(SESSION_NAME).block(VERIFY_TIMEOUT); + assertNotNull(replacement); + Assertions.assertNotSame(first, replacement); + assertFalse(connection.removeSession(SESSION_NAME, first)); + assertFalse(replacement.isDisposed()); + assertSame(replacement, connection.createSession(SESSION_NAME).block(VERIFY_TIMEOUT)); + assertTrue(connection.removeSession(SESSION_NAME, replacement)); + assertFalse(connection.removeSession(SESSION_NAME, replacement)); + } + /** * Verifies initial endpoint state is uninitialized and completes when the connection is closed. */ 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 f05c9c4b9b208..1b4f51ab58173 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 @@ -42,6 +42,7 @@ import org.mockito.Mock; import org.mockito.Mockito; import org.mockito.MockitoAnnotations; +import reactor.core.Exceptions; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.scheduler.Scheduler; @@ -186,6 +187,21 @@ public void testLinkSize() { verify(sender).getRemoteMaxMessageSize(); } + @Test + public void linkSizeNegotiationPreservesDetachAsRetryExhaustionCause() { + options.setMaxRetries(0); + reactorSender = spy(new ReactorSender(amqpConnection, ENTITY_PATH, sender, handler, reactorProvider, + tokenManager, messageSerializer, options, scheduler, AmqpMetricsProvider.noop())); + final AmqpException detach + = new AmqpException(true, AmqpErrorCondition.LINK_DETACH_FORCED, "detached during negotiation", null); + doReturn(Flux.error(detach).cache(1)).when(reactorSender).getEndpointStates(); + + StepVerifier.create(reactorSender.getLinkSize()).expectErrorSatisfies(error -> { + assertTrue(Exceptions.isRetryExhausted(error)); + Assertions.assertSame(detach, error.getCause()); + }).verify(VERIFY_TIMEOUT); + } + /** * Verifies that getMaxBatchSize reads the vendor property com.microsoft:max-message-batch-size from the link's * remote properties and returns its value. diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSessionCacheTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSessionCacheTest.java index 4228109da04c9..e62c9eec91f95 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSessionCacheTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSessionCacheTest.java @@ -210,6 +210,50 @@ void shouldNotEvictSessionIfOwnerDisposed() { .verifyComplete(); } + @Test + void shouldEvictOnlyExpectedSession() { + final ReactorSessionCache cache = createSessionCache(); + final Connection connection = mock(Connection.class); + final String name = "session-0"; + final ReactorSession first = session(name, endpointStatesSink(), null); + final ReactorSession replacement = session(name, endpointStatesSink(), null); + final HashMap> lookup = new HashMap<>(); + lookup.put(name, sessions(first, replacement)); + final CacheLoader loader = new CacheLoader(lookup); + Assertions.assertSame(first, cache.getOrLoad(Mono.just(connection), name, loader).block()); + Assertions.assertFalse(cache.evict(name, replacement)); + Assertions.assertTrue(cache.evict(name, first)); + Assertions.assertSame(replacement, cache.getOrLoad(Mono.just(connection), name, loader).block()); + Assertions.assertFalse(cache.evict(name, first)); + Assertions.assertSame(replacement, cache.getOrLoad(Mono.just(connection), name, loader).block()); + Mockito.verify(replacement, Mockito.never()).closeAsync(anyString(), any(), eq(true)); + Assertions.assertTrue(cache.evict(name, replacement)); + Assertions.assertFalse(cache.evict(name, replacement)); + } + + @Test + void delayedOpenFailureDoesNotEvictReplacementSession() { + final ReactorSessionCache cache = createSessionCache(); + final Connection connection = mock(Connection.class); + final String name = "session-0"; + final ReactorSession first = session(name, endpointStatesSink(), null); + final ReactorSession replacement = session(name, endpointStatesSink(), null); + final Sinks.One opening = Sinks.one(); + when(first.open()).thenReturn(opening.asMono()); + final HashMap> lookup = new HashMap<>(); + lookup.put(name, sessions(first, replacement)); + final CacheLoader loader = new CacheLoader(lookup); + final RuntimeException failure = new RuntimeException("delayed open failure"); + StepVerifier.create(cache.getOrLoad(Mono.just(connection), name, loader)).then(() -> { + Assertions.assertTrue(cache.evict(name)); + Assertions.assertSame(replacement, cache.getOrLoad(Mono.just(connection), name, loader).block()); + opening.emitError(failure, Sinks.EmitFailureHandler.FAIL_FAST); + }).verifyErrorMatches(error -> error == failure); + Assertions.assertSame(replacement, cache.getOrLoad(Mono.just(connection), name, loader).block()); + Mockito.verify(replacement, Mockito.never()).closeAsync(anyString(), any(), eq(true)); + Assertions.assertTrue(cache.evict(name, replacement)); + } + private ReactorSessionCache createSessionCache() { return new ReactorSessionCache(CONNECTION_ID, NAMESPACE, handlerProvider, reactorProvider, OPEN_TIMEOUT, logger); diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RecoveryKindTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RecoveryKindTest.java new file mode 100644 index 0000000000000..a3ed21a9eacf1 --- /dev/null +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RecoveryKindTest.java @@ -0,0 +1,231 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.amqp.implementation; + +import com.azure.core.amqp.exception.AmqpErrorCondition; +import com.azure.core.amqp.exception.AmqpException; +import com.azure.core.amqp.exception.OperationCancelledException; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.TimeoutException; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Tests for {@link RecoveryKind#classify(Throwable)}. + */ +class RecoveryKindTest { + + @Test + void nullErrorReturnsNone() { + assertEquals(RecoveryKind.NONE, RecoveryKind.classify(null)); + } + + @Test + void timeoutExceptionReturnsNone() { + assertEquals(RecoveryKind.NONE, RecoveryKind.classify(new TimeoutException("timed out"))); + } + + @Test + void serverBusyReturnsNone() { + final AmqpException error = new AmqpException(true, AmqpErrorCondition.SERVER_BUSY_ERROR, "server busy", null); + assertEquals(RecoveryKind.NONE, RecoveryKind.classify(error)); + } + + @Test + void timeoutErrorConditionReturnsNone() { + final AmqpException error = new AmqpException(true, AmqpErrorCondition.TIMEOUT_ERROR, "timeout", null); + assertEquals(RecoveryKind.NONE, RecoveryKind.classify(error)); + } + + @Test + void linkDetachForcedReturnsLink() { + final AmqpException error + = new AmqpException(true, AmqpErrorCondition.LINK_DETACH_FORCED, "detach forced", null); + assertEquals(RecoveryKind.LINK, RecoveryKind.classify(error)); + } + + @Test + void linkStolenReturnsLink() { + final AmqpException error = new AmqpException(true, AmqpErrorCondition.LINK_STOLEN, "link stolen", null); + assertEquals(RecoveryKind.LINK, RecoveryKind.classify(error)); + } + + @Test + void transientAmqpErrorWithoutConditionReturnsLink() { + final AmqpException error = new AmqpException(true, "transient error", null, null); + assertEquals(RecoveryKind.LINK, RecoveryKind.classify(error)); + } + + @Test + void connectionForcedReturnsConnection() { + final AmqpException error + = new AmqpException(true, AmqpErrorCondition.CONNECTION_FORCED, "connection forced", null); + assertEquals(RecoveryKind.CONNECTION, RecoveryKind.classify(error)); + } + + @Test + void connectionFramingErrorReturnsConnection() { + final AmqpException error + = new AmqpException(true, AmqpErrorCondition.CONNECTION_FRAMING_ERROR, "framing error", null); + assertEquals(RecoveryKind.CONNECTION, RecoveryKind.classify(error)); + } + + @Test + void internalErrorReturnsConnection() { + final AmqpException error = new AmqpException(true, AmqpErrorCondition.INTERNAL_ERROR, "internal error", null); + assertEquals(RecoveryKind.CONNECTION, RecoveryKind.classify(error)); + } + + @Test + void protonIoReturnsConnection() { + final AmqpException error = new AmqpException(true, AmqpErrorCondition.PROTON_IO, "io error", null); + assertEquals(RecoveryKind.CONNECTION, RecoveryKind.classify(error)); + } + + @Test + void connectionRedirectReturnsConnection() { + final AmqpException error = new AmqpException(true, AmqpErrorCondition.CONNECTION_REDIRECT, "redirect", null); + assertEquals(RecoveryKind.CONNECTION, RecoveryKind.classify(error)); + } + + @Test + void linkRedirectReturnsLink() { + final AmqpException error = new AmqpException(true, AmqpErrorCondition.LINK_REDIRECT, "redirect", null); + assertEquals(RecoveryKind.LINK, RecoveryKind.classify(error)); + } + + @Test + void transferLimitExceededReturnsLink() { + final AmqpException error + = new AmqpException(true, AmqpErrorCondition.TRANSFER_LIMIT_EXCEEDED, "transfer limit", null); + assertEquals(RecoveryKind.LINK, RecoveryKind.classify(error)); + } + + @Test + void argumentErrorReturnsFatal() { + final AmqpException error = new AmqpException(false, AmqpErrorCondition.ARGUMENT_ERROR, "bad argument", null); + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(error)); + } + + @Test + void notFoundReturnsFatal() { + final AmqpException error = new AmqpException(false, AmqpErrorCondition.NOT_FOUND, "not found", null); + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(error)); + } + + @Test + void unauthorizedAccessReturnsFatal() { + final AmqpException error + = new AmqpException(false, AmqpErrorCondition.UNAUTHORIZED_ACCESS, "unauthorized", null); + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(error)); + } + + @Test + void payloadSizeExceededReturnsFatal() { + final AmqpException error + = new AmqpException(false, AmqpErrorCondition.LINK_PAYLOAD_SIZE_EXCEEDED, "too large", null); + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(error)); + } + + @Test + void notAllowedReturnsFatal() { + final AmqpException error = new AmqpException(false, AmqpErrorCondition.NOT_ALLOWED, "not allowed", null); + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(error)); + } + + @Test + void nonTransientAmqpErrorReturnsFatal() { + final AmqpException error = new AmqpException(false, "permanent error", null, null); + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(error)); + } + + @Test + void requestResponseChannelClosedReturnsLink() { + final RequestResponseChannelClosedException error = new RequestResponseChannelClosedException("channel closed"); + assertEquals(RecoveryKind.LINK, RecoveryKind.classify(error)); + } + + @Test + void unknownExceptionReturnsFatal() { + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(new RuntimeException("unknown"))); + } + + @Test + void sessionLockLostReturnsLink() { + final AmqpException error + = new AmqpException(true, AmqpErrorCondition.SESSION_LOCK_LOST, "session lock lost", null); + assertEquals(RecoveryKind.LINK, RecoveryKind.classify(error)); + } + + @Test + void messageLockLostReturnsFatal() { + final AmqpException error + = new AmqpException(false, AmqpErrorCondition.MESSAGE_LOCK_LOST, "message lock lost", null); + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(error)); + } + + @Test + void storeLockLostReturnsFatal() { + final AmqpException error + = new AmqpException(false, AmqpErrorCondition.STORE_LOCK_LOST_ERROR, "store lock lost", null); + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(error)); + } + + @Test + void operationCancelledReturnsLink() { + final AmqpException error = new AmqpException(true, AmqpErrorCondition.OPERATION_CANCELLED, "cancelled", null); + assertEquals(RecoveryKind.LINK, RecoveryKind.classify(error)); + } + + @Test + void abortedSendWithoutErrorConditionReturnsLink() { + assertEquals(RecoveryKind.LINK, RecoveryKind.classify(new OperationCancelledException("Released", null))); + assertEquals(RecoveryKind.LINK, RecoveryKind.classify( + new OperationCancelledException("Delivery advance failed", new IllegalStateException("advance"), null))); + } + + @Test + void lockedNamedSessionReturnsFatal() { + final AmqpException error = new AmqpException(false, AmqpErrorCondition.SESSION_CANNOT_BE_LOCKED, + "The requested session is already locked.", null); + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(error)); + } + + @Test + void resourceLimitExceededReturnsNone() { + final AmqpException error + = new AmqpException(true, AmqpErrorCondition.RESOURCE_LIMIT_EXCEEDED, "resource limit", null); + assertEquals(RecoveryKind.NONE, RecoveryKind.classify(error)); + } + + @Test + void illegalStateExceptionDisposedMessageReturnsLink() { + // Matches ReactorSender.send() message: "connectionId[%s] linkName[%s] Cannot publish message when disposed." + assertEquals(RecoveryKind.LINK, RecoveryKind.classify( + new IllegalStateException("connectionId[abc] linkName[xyz] Cannot publish message when disposed."))); + } + + @Test + void illegalStateExceptionDisposedDataBatchReturnsLink() { + // Matches ReactorSender.send(List) message: + // "connectionId[%s] linkName[%s] Cannot publish data batch when disposed." + assertEquals(RecoveryKind.LINK, RecoveryKind.classify( + new IllegalStateException("connectionId[abc] linkName[xyz] Cannot publish data batch when disposed."))); + } + + @Test + void illegalStateExceptionUnrelatedToDisposedReturnsFatal() { + // Non-disposed IllegalStateException must remain FATAL (genuine application or SDK bug). + assertEquals(RecoveryKind.FATAL, RecoveryKind.classify(new IllegalStateException("some unexpected state"))); + } + + @Test + void illegalStateExceptionConnectionDisposedReturnsFatal() { + // "Connection is disposed. Cannot get management instance." contains "disposed" but NOT + // "Cannot publish" — must remain FATAL to avoid misclassifying connection-level disposal. + assertEquals(RecoveryKind.FATAL, RecoveryKind + .classify(new IllegalStateException("Connection is disposed. Cannot get management instance."))); + } +} diff --git a/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxHttpClientExpectContinueTests.java b/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxHttpClientExpectContinueTests.java index 427c5696a229c..eedc2d3b15237 100644 --- a/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxHttpClientExpectContinueTests.java +++ b/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxHttpClientExpectContinueTests.java @@ -8,6 +8,8 @@ import com.azure.core.http.HttpMethod; import com.azure.core.http.HttpRequest; import com.azure.core.http.HttpResponse; +import io.vertx.core.Vertx; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -43,6 +45,29 @@ public class VertxHttpClientExpectContinueTests { // Keep this below the server's 500 ms settle window and the one-second continue fallback. private static final Duration WRITE_TIMEOUT = Duration.ofMillis(250); + @BeforeAll + public static void warmUpHttpClient() throws Exception { + // Netty 4.2 initializes its allocation JFR event on the first header write. Keep that cold-start work + // outside the short write deadline used below to distinguish header writes from waiting for Continue. + Vertx vertx = Vertx.vertx(); + try (ExpectContinueServer server = new ExpectContinueServer()) { + server.start(); + HttpClient client = new VertxHttpClientBuilder().vertx(vertx) + .writeTimeout(Duration.ZERO) + .responseTimeout(RESPONSE_TIMEOUT) + .build(); + HttpRequest request = new HttpRequest(HttpMethod.PUT, server.url()).setBody(BODY); + request.getHeaders().set(HttpHeaderName.EXPECT, "100-continue"); + try (HttpResponse response = client.send(request).block(TEST_TIMEOUT)) { + assertNotNull(response); + assertEquals(201, response.getStatusCode()); + } + assertTrue(server.awaitRequest(), "the warm-up request never reached the server"); + } finally { + vertx.close().toCompletionStage().toCompletableFuture().get(TEST_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + } + } + @ParameterizedTest @ValueSource(strings = { "100-continue", "100-Continue", " 100-continue ", "100-continue, foo" }) public void withholdsBodyUntilContinue(String expectHeaderValue) throws Exception { diff --git a/sdk/servicebus/azure-messaging-servicebus/pom.xml b/sdk/servicebus/azure-messaging-servicebus/pom.xml index c1e8fa6c21f88..aeb5d840d4a01 100644 --- a/sdk/servicebus/azure-messaging-servicebus/pom.xml +++ b/sdk/servicebus/azure-messaging-servicebus/pom.xml @@ -44,6 +44,7 @@ --add-exports com.azure.core/com.azure.core.implementation.util=ALL-UNNAMED --add-opens com.azure.core/com.azure.core.implementation.util=ALL-UNNAMED + --add-opens com.azure.core.amqp/com.azure.core.amqp.implementation=ALL-UNNAMED --add-reads com.azure.messaging.servicebus=com.azure.http.netty --add-reads com.azure.messaging.servicebus=com.azure.core.experimental --add-reads com.azure.messaging.servicebus=com.azure.core.tracing.opentelemetry @@ -69,7 +70,7 @@ com.azure azure-core-amqp - 2.13.0 + 2.14.0-beta.1 com.azure diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ConnectionCacheWrapper.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ConnectionCacheWrapper.java index 1d4aa44b4c422..870b3f74b4b24 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ConnectionCacheWrapper.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ConnectionCacheWrapper.java @@ -51,4 +51,18 @@ AmqpRetryOptions getRetryOptions() { boolean isChannelClosed() { return isV2 ? cache.isCurrentConnectionClosed() : processor.isChannelClosed(); } + + /** + * Invalidates the connection used by the failed operation without disturbing a replacement. + * On v2, marks that connection for eviction on the next get(). On v1, closes it immediately. + * + * @param connection The connection used by the failed operation. + */ + void invalidateConnection(ServiceBusAmqpConnection connection) { + if (isV2) { + cache.invalidateConnection(connection.getId()); + } else { + processor.forceCloseChannel(connection); + } + } } diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/RecoveryUtils.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/RecoveryUtils.java new file mode 100644 index 0000000000000..f3a108e89fdbf --- /dev/null +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/RecoveryUtils.java @@ -0,0 +1,59 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.servicebus; + +import com.azure.core.amqp.exception.AmqpErrorCondition; +import com.azure.core.amqp.exception.AmqpException; +import com.azure.core.amqp.implementation.RecoveryKind; +import reactor.core.Exceptions; + +import java.util.concurrent.TimeoutException; + +/** + * Package-private utilities shared by the sender and receiver call sites that compose + * tiered recovery on top of {@link com.azure.core.amqp.implementation.RetryUtil#withRetry}. + */ +final class RecoveryUtils { + + private RecoveryUtils() { + } + + static Throwable unwrapRetryExhaustion(Throwable error) { + while (Exceptions.isRetryExhausted(error) && error.getCause() != null) { + error = error.getCause(); + } + return error; + } + + /** + * Ensures the error is retriable by the standard + * {@link com.azure.core.amqp.implementation.RetryUtil#createRetry} filter. + * That filter only accepts {@link TimeoutException} or transient {@link AmqpException}. + * Non-AMQP errors like {@code IllegalStateException("Cannot publish when disposed")} + * that {@link RecoveryKind} classifies as LINK would otherwise be rejected, bypassing + * the recovery just performed. This wraps such errors as transient {@link AmqpException} + * so the same recovery + retry composition works for both sender and receiver paths. + * + * @param error the original error emitted from a send/receive operation. + * @return the same error if it already passes the retry filter, otherwise a transient + * {@link AmqpException} wrapping it for LINK/CONNECTION errors or a same-link resource-limit retry. + */ + static Throwable asRetriable(Throwable error) { + if (error instanceof TimeoutException + || (error instanceof AmqpException && ((AmqpException) error).isTransient())) { + return error; + } + if (error instanceof AmqpException) { + final AmqpException amqpError = (AmqpException) error; + if (amqpError.getErrorCondition() == AmqpErrorCondition.RESOURCE_LIMIT_EXCEEDED) { + return new AmqpException(true, amqpError.getErrorCondition(), amqpError, amqpError.getContext()); + } + } + final RecoveryKind kind = RecoveryKind.classify(error); + if (kind == RecoveryKind.LINK || kind == RecoveryKind.CONNECTION) { + return new AmqpException(true, error.getMessage(), error, null); + } + return error; + } +} diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java index 0b3154ec33efb..07d4cb8566ba6 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java @@ -11,6 +11,7 @@ import com.azure.core.amqp.implementation.MessageFlux; import com.azure.core.amqp.implementation.MessageSerializer; import com.azure.core.amqp.implementation.RequestResponseChannelClosedException; +import com.azure.core.amqp.implementation.RecoveryKind; import com.azure.core.amqp.implementation.RetryUtil; import com.azure.core.amqp.implementation.StringUtil; import com.azure.core.amqp.implementation.handler.DeliveryNotOnLinkException; @@ -1713,9 +1714,14 @@ private ServiceBusAsyncConsumer getOrCreateConsumer() { // The Mono, when subscribed, creates a ServiceBusReceiveLink in the ServiceBusAmqpConnection emitted by the connectionProcessor // - final Mono receiveLinkMono = connectionProcessor.flatMap(connection -> { - return connection.createReceiveLink(linkName, entityPath, receiverOptions.getReceiveMode(), null, - entityType, identifier); + final AtomicReference operationConnection = new AtomicReference<>(); + final Mono receiveLinkMono = Mono.defer(() -> { + operationConnection.set(null); + return connectionProcessor.flatMap(connection -> { + operationConnection.set(connection); + return connection.createReceiveLink(linkName, entityPath, receiverOptions.getReceiveMode(), null, + entityType, identifier); + }); }).doOnNext(next -> { LOGGER.atVerbose() .addKeyValue(LINK_NAME_KEY, linkName) @@ -1736,14 +1742,40 @@ private ServiceBusAsyncConsumer getOrCreateConsumer() { // final Mono retryableReceiveLinkMono = RetryUtil.withRetry(receiveLinkMono.onErrorMap(RequestResponseChannelClosedException.class, e -> { - // When the current connection is being disposed, the V1 ConnectionProcessor or V2 ReactorConnectionCache - // can produce a new connection if downstream request. In this context, treat - // RequestResponseChannelClosedException error from the following two sources as retry-able so that - // retry can obtain a new connection - - // 1. error from the RequestResponseChannel scoped to the current connection being disposed, - // 2. error from the V2 RequestResponseChannelCache scoped to the current connection being disposed. - // return new AmqpException(true, e.getMessage(), e, null); + }).onErrorResume(e -> { + final RecoveryKind recoveryKind = RecoveryKind.classify(e); + if (recoveryKind == RecoveryKind.LINK || recoveryKind == RecoveryKind.CONNECTION) { + LOGGER.atWarning() + .addKeyValue(LINK_NAME_KEY, linkName) + .addKeyValue("recoveryKind", recoveryKind) + .log("Receive link creation failed, performing {} recovery.", recoveryKind, e); + + // The receive-link factory evicts only the session that hosted the failed link. + // CONNECTION recovery also invalidates the connection used by this attempt. + final ServiceBusAmqpConnection failedConnection = operationConnection.getAndSet(null); + Mono recovery = Mono.fromRunnable(() -> { + if (failedConnection == null) { + return; + } + if (recoveryKind == RecoveryKind.CONNECTION) { + connectionCacheWrapper.invalidateConnection(failedConnection); + } + }).onErrorResume(error -> { + LOGGER.atWarning() + .addKeyValue(LINK_NAME_KEY, linkName) + .log("Error invalidating connection during {} recovery.", recoveryKind, error); + return Mono.empty(); + }); + + // Use the shared retry-filter wrapper so non-AMQP errors classified as + // LINK/CONNECTION (e.g., IllegalStateException) are wrapped as transient + // AmqpException and accepted by RetryUtil.createRetry. Centralising this + // logic in RecoveryUtils keeps the sender and receiver paths consistent. + final Throwable retriable = RecoveryUtils.asRetriable(e); + return recovery.then(Mono.error(retriable)); + } + return Mono.error(RecoveryUtils.asRetriable(e)); }), connectionCacheWrapper.getRetryOptions(), "Failed to create receive link " + linkName, true); // A Flux that produces a new AmqpReceiveLink each time it receives a request from the below diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClient.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClient.java index 260b2f38e29de..ee18d049231d4 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClient.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClient.java @@ -10,7 +10,9 @@ import com.azure.core.amqp.implementation.AmqpSendLink; import com.azure.core.amqp.implementation.ErrorContextProvider; import com.azure.core.amqp.implementation.MessageSerializer; +import com.azure.core.amqp.implementation.RecoveryKind; import com.azure.core.amqp.implementation.RequestResponseChannelClosedException; +import com.azure.core.amqp.implementation.RetryUtil; import com.azure.core.annotation.ServiceClient; import com.azure.core.util.CoreUtils; import com.azure.core.util.logging.ClientLogger; @@ -24,6 +26,8 @@ import org.apache.qpid.proton.amqp.transaction.TransactionalState; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.util.function.Tuple2; +import reactor.util.function.Tuples; import java.time.OffsetDateTime; import java.util.ArrayList; @@ -44,7 +48,6 @@ import java.util.stream.Collector; import static com.azure.core.amqp.implementation.ClientConstants.ENTITY_PATH_KEY; -import static com.azure.core.amqp.implementation.RetryUtil.withRetry; import static com.azure.core.util.FluxUtil.fluxError; import static com.azure.core.util.FluxUtil.monoError; import static com.azure.messaging.servicebus.implementation.Messages.INVALID_OPERATION_DISPOSED_SENDER; @@ -240,6 +243,7 @@ public final class ServiceBusSenderAsyncClient implements AutoCloseable { private final MessagingEntityType entityType; private final Runnable onClientClose; private final String entityName; + private final ConnectionCacheWrapper connectionCacheWrapper; private final Mono connectionProcessor; private final String fullyQualifiedNamespace; private final String viaEntityName; @@ -260,6 +264,7 @@ public final class ServiceBusSenderAsyncClient implements AutoCloseable { this.retryOptions = Objects.requireNonNull(retryOptions, "'retryOptions' cannot be null."); this.entityName = Objects.requireNonNull(entityName, "'entityPath' cannot be null."); Objects.requireNonNull(connectionCacheWrapper, "'connectionCacheWrapper' cannot be null."); + this.connectionCacheWrapper = connectionCacheWrapper; this.connectionProcessor = connectionCacheWrapper.getConnection(); this.fullyQualifiedNamespace = connectionCacheWrapper.getFullyQualifiedNamespace(); this.instrumentation = Objects.requireNonNull(instrumentation, "'instrumentation' cannot be null."); @@ -468,24 +473,24 @@ public Mono createMessageBatch(CreateMessageBatchOptions final int maxSize = options.getMaximumSizeInBytes(); - return getSendLinkWithRetry("create-batch") - .flatMap(link -> link.getMaxBatchSize().flatMap(batchSizeFromLink -> { - // Use the value from getMaxBatchSize() (vendor property, or standard max-message-size fallback - // in ReactorSender). If neither is available (batchSizeFromLink <= 0), use 1 MB as a - // last-resort default to prevent oversized batches on broken links. - final int maximumBatchSize = batchSizeFromLink > 0 ? batchSizeFromLink : DEFAULT_MAX_BATCH_SIZE_BYTES; - if (maxSize > maximumBatchSize) { - return monoError(logger, - new IllegalArgumentException(String.format(Locale.US, - "CreateMessageBatchOptions.getMaximumSizeInBytes (%s bytes) is larger than the maximum" - + " batch size (%s bytes).", - maxSize, maximumBatchSize))); - } - final int batchSize = maxSize > 0 ? maxSize : maximumBatchSize; - return Mono.just( - new ServiceBusMessageBatch(isV2, batchSize, link::getErrorContext, tracer, messageSerializer)); - })) - .onErrorMap(this::mapError); + return getSendLinkAndMaxBatchSizeWithRetry("create-batch").flatMap(t -> { + final AmqpSendLink link = t.getT1(); + final int batchSizeFromLink = t.getT2(); + // Use the value from getMaxBatchSize() (vendor property, or standard max-message-size fallback + // in ReactorSender). If neither is available (batchSizeFromLink <= 0), use 1 MB as a + // last-resort default to prevent oversized batches on broken links. + final int maximumBatchSize = batchSizeFromLink > 0 ? batchSizeFromLink : DEFAULT_MAX_BATCH_SIZE_BYTES; + if (maxSize > maximumBatchSize) { + return monoError(logger, + new IllegalArgumentException(String.format(Locale.US, + "CreateMessageBatchOptions.getMaximumSizeInBytes (%s bytes) is larger than the maximum" + + " batch size (%s bytes).", + maxSize, maximumBatchSize))); + } + final int batchSize = maxSize > 0 ? maxSize : maximumBatchSize; + return Mono + .just(new ServiceBusMessageBatch(isV2, batchSize, link::getErrorContext, tracer, messageSerializer)); + }).onErrorMap(this::mapError); } /** @@ -822,7 +827,9 @@ private Mono scheduleMessageInternal(ServiceBusMessage message, OffsetDate } return tracer.traceScheduleMono("ServiceBus.scheduleMessage", - getSendLinkWithRetry("schedule-message").flatMap(link -> link.getLinkSize().flatMap(size -> { + getSendLinkAndSizeWithRetry("schedule-message").flatMap(t -> { + final AmqpSendLink link = t.getT1(); + final int size = t.getT2(); final int maxSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES; return connectionProcessor.flatMap(connection -> connection.getManagementNode(entityName, entityType)) .flatMap( @@ -830,7 +837,7 @@ private Mono scheduleMessageInternal(ServiceBusMessage message, OffsetDate .schedule(Arrays.asList(message), scheduledEnqueueTime, maxSize, link.getLinkName(), transactionContext) .next()); - })), message, message.getContext()).onErrorMap(this::mapError); + }), message, message.getContext()).onErrorMap(this::mapError); } /** @@ -875,20 +882,28 @@ private Mono sendBatchInternal(ServiceBusMessageBatch batch, ServiceBusTra messages.add(message); }); - final Mono sendMessage = getSendLink("send-batch").flatMap(link -> { - if (transactionContext != null && transactionContext.getTransactionId() != null) { - final TransactionalState deliveryState = new TransactionalState(); - deliveryState.setTxnId(Binary.create(transactionContext.getTransactionId())); - return messages.size() == 1 - ? link.send(messages.get(0), deliveryState) - : link.send(messages, deliveryState); - } else { - return messages.size() == 1 ? link.send(messages.get(0)) : link.send(messages); - } - }); + final Mono withRetry = Mono.defer(() -> { + final AtomicReference operationLink = new AtomicReference<>(); + final AtomicReference operationConnection = new AtomicReference<>(); + final Mono sendMessage = getSendLink("send-batch", operationConnection).flatMap(link -> { + operationLink.set(link); + if (transactionContext != null && transactionContext.getTransactionId() != null) { + final TransactionalState deliveryState = new TransactionalState(); + deliveryState.setTxnId(Binary.create(transactionContext.getTransactionId())); + return messages.size() == 1 + ? link.send(messages.get(0), deliveryState) + : link.send(messages, deliveryState); + } else { + return messages.size() == 1 ? link.send(messages.get(0)) : link.send(messages); + } + }); - final String message = "Sending messages timed out. message-count:" + batch.getCount() + entityId(); - final Mono withRetry = withRetry(sendMessage, retryOptions, message).onErrorMap(this::mapError); + final String timeoutMessage = "Sending messages timed out. message-count:" + batch.getCount() + entityId(); + return RetryUtil.withRetry(sendMessage.onErrorMap(RecoveryUtils::unwrapRetryExhaustion) + .onErrorResume(e -> recoverBeforeRetry(e, "sendBatch", operationLink, operationConnection) + .then(Mono.error(RecoveryUtils.asRetriable(e)))), + retryOptions, timeoutMessage); + }).onErrorMap(this::mapError); // The single-message send path (sendMessageInternal) starts the "ServiceBus.send" span itself on the // subscribing thread, so it sends this batch with instrument=false to avoid a duplicate span. if (!instrument) { @@ -905,19 +920,27 @@ private Mono sendMessageInternal(ServiceBusMessage message, ServiceBusTran final List messageList = Collections.singletonList(message); + // Apply retry+recovery to BOTH link acquisition and link-size negotiation. Keeping the + // batch collection outside the retry boundary avoids re-subscribing on each retry attempt. + // A getLinkSize() failure (link never ACTIVE, link disposed during negotiation) now triggers + // the same tiered recovery as a getSendLink() failure. + // // Uses getLinkSize() intentionally — this path is for single-message sends only (sendMessage()). // Single messages should use the full link capacity (up to 100 MB on Premium large-message), // not the batch-size cap. The batch path (sendMessages(Iterable), scheduleMessages(Iterable)) // goes through createMessageBatch() which calls getMaxBatchSize() for the vendor-property-based cap. - final Mono> batchList - = getSendLinkWithRetry("send-batches").flatMap(link -> link.getLinkSize().flatMap(size -> { - final int batchSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES; - final CreateMessageBatchOptions batchOptions - = new CreateMessageBatchOptions().setMaximumSizeInBytes(batchSize); - return Flux.fromIterable(messageList) - .collect(new AmqpMessageCollector(isV2, batchOptions, 1, link::getErrorContext, tracer, - messageSerializer)); - })); + final Mono> linkAndSize = getSendLinkAndSizeWithRetry("send-batches"); + + final Mono> batchList = linkAndSize.flatMap(t -> { + final AmqpSendLink link = t.getT1(); + final int size = t.getT2(); + final int batchSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES; + final CreateMessageBatchOptions batchOptions + = new CreateMessageBatchOptions().setMaximumSizeInBytes(batchSize); + return Flux.fromIterable(messageList) + .collect( + new AmqpMessageCollector(isV2, batchOptions, 1, link::getErrorContext, tracer, messageSerializer)); + }); // The raw send pipeline, without span instrumentation. Instrumentation is applied as the // outermost operator below so the span is created on the subscribing thread (instrument=false here). @@ -937,14 +960,19 @@ private Mono sendMessageInternal(ServiceBusMessage message, ServiceBusTran }); } - private Mono getSendLink(String callSite) { - final Mono getSendLink = connectionProcessor.flatMap(connection -> { - if (!CoreUtils.isNullOrEmpty(viaEntityName)) { - return connection.createSendLink("VIA-".concat(viaEntityName), viaEntityName, retryOptions, entityName, - identifier); - } else { - return connection.createSendLink(entityName, entityName, retryOptions, null, identifier); - } + private Mono getSendLink(String callSite, + AtomicReference operationConnection) { + final Mono getSendLink = Mono.defer(() -> { + operationConnection.set(null); + return connectionProcessor.flatMap(connection -> { + operationConnection.set(connection); + if (!CoreUtils.isNullOrEmpty(viaEntityName)) { + return connection.createSendLink("VIA-".concat(viaEntityName), viaEntityName, retryOptions, + entityName, identifier); + } else { + return connection.createSendLink(entityName, entityName, retryOptions, null, identifier); + } + }); }) .doOnNext(link -> linkName.compareAndSet(null, link.getLinkName())) .onErrorMap(RequestResponseChannelClosedException.class, e -> { @@ -964,8 +992,55 @@ private Mono getSendLink(String callSite) { return getSendLink; } - private Mono getSendLinkWithRetry(String callSite) { - return withRetry(getSendLink(callSite), retryOptions, String.format(retryGetLinkErrorMessageFormat, callSite)); + /** + * Acquires a send link AND its negotiated maximum message size inside a single retry+recovery + * boundary. A failure during {@code getLinkSize()} (e.g., the link never reaches ACTIVE within + * the timeout, or the link becomes disposed mid-negotiation) triggers the same tiered recovery + * as a failure during {@code getSendLink()} — without it, transient size-negotiation faults + * would short-circuit the operation without LINK or CONNECTION recovery. + * + * @param callSite identifier used in recovery diagnostics. + * @return a Mono emitting (link, size) on success, retried with recovery on transient failures. + */ + private Mono> getSendLinkAndSizeWithRetry(String callSite) { + return Mono.defer(() -> { + final AtomicReference operationLink = new AtomicReference<>(); + final AtomicReference operationConnection = new AtomicReference<>(); + return RetryUtil + .withRetry( + getSendLink(callSite, operationConnection).doOnNext(operationLink::set) + .flatMap(link -> link.getLinkSize().map(size -> Tuples.of(link, size))) + .onErrorMap(RecoveryUtils::unwrapRetryExhaustion) + .onErrorResume(e -> recoverBeforeRetry(e, "getSendLinkAndSize-" + callSite, operationLink, + operationConnection).then(Mono.error(RecoveryUtils.asRetriable(e)))), + retryOptions, String.format(retryGetLinkErrorMessageFormat, callSite)); + }); + } + + /** + * Acquires a send link AND its negotiated maximum batch size inside a single retry+recovery + * boundary. Used by {@code createMessageBatch()} which caps batch size at the vendor-property + * limit (see {@link #DEFAULT_MAX_BATCH_SIZE_BYTES}) rather than full link capacity. A failure + * during {@code getMaxBatchSize()} (e.g., the link never reaches ACTIVE within the timeout, or + * the link becomes disposed mid-negotiation) triggers the same tiered recovery as a failure + * during {@code getSendLink()} — without it, transient size-negotiation faults would + * short-circuit the operation without LINK or CONNECTION recovery. + * + * @param callSite identifier used in recovery diagnostics. + * @return a Mono emitting (link, batchSize) on success, retried with recovery on transient failures. + */ + private Mono> getSendLinkAndMaxBatchSizeWithRetry(String callSite) { + return Mono.defer(() -> { + final AtomicReference operationLink = new AtomicReference<>(); + final AtomicReference operationConnection = new AtomicReference<>(); + return RetryUtil.withRetry( + getSendLink(callSite, operationConnection).doOnNext(operationLink::set) + .flatMap(link -> link.getMaxBatchSize().map(size -> Tuples.of(link, size))) + .onErrorMap(RecoveryUtils::unwrapRetryExhaustion) + .onErrorResume(e -> recoverBeforeRetry(e, "getSendLinkAndMaxBatchSize-" + callSite, operationLink, + operationConnection).then(Mono.error(RecoveryUtils.asRetriable(e)))), + retryOptions, String.format(retryGetLinkErrorMessageFormat, callSite)); + }); } private Throwable mapError(Throwable throwable) { @@ -975,6 +1050,59 @@ private Throwable mapError(Throwable throwable) { return throwable; } + /** + * Performs reactive tiered recovery based on the classified error. For LINK recovery, closes the + * send link so the next retry creates a fresh one. For CONNECTION recovery, invalidates the + * operation's connection before awaiting link closure so a stalled close cannot prevent eviction. + * + *

Called from {@code onErrorResume} at each send call site so that recovery completes before + * the error is re-emitted and the standard retry logic re-subscribes.

+ */ + private Mono recoverBeforeRetry(Throwable error, String callSite, AtomicReference linkRef, + AtomicReference connectionRef) { + final RecoveryKind kind = RecoveryKind.classify(error); + if (kind != RecoveryKind.LINK && kind != RecoveryKind.CONNECTION) { + return Mono.empty(); + } + + logger.atWarning() + .addKeyValue(ENTITY_PATH_KEY, entityName) + .addKeyValue("recoveryKind", kind) + .addKeyValue("callSite", callSite) + .log("Performing {} recovery before retry.", kind, error); + + // Close the operation-scoped send link so the next retry creates a fresh one. + // Using a per-operation AtomicReference (not a class-level field) prevents concurrent send + // operations from accidentally closing each other's links. + // Use closeAsync() rather than dispose() to avoid blocking the Reactor thread. + Mono recovery = Mono.empty(); + final AmqpSendLink link = linkRef != null ? linkRef.getAndSet(null) : null; + if (link != null) { + recovery = Mono.defer(() -> { + Mono close = link.closeAsync(); + return close != null ? close : Mono.empty(); + }).onErrorResume(closeErr -> { + logger.atVerbose() + .addKeyValue(ENTITY_PATH_KEY, entityName) + .log("Error closing stale send link during recovery.", closeErr); + return Mono.empty(); + }); + } + linkName.set(null); + + // For CONNECTION errors, invalidate the cached connection so the next get() closes it + // and creates a fresh one. Matches Go SDK's Namespace.Recover(). + if (kind == RecoveryKind.CONNECTION) { + final ServiceBusAmqpConnection failedConnection = connectionRef.getAndSet(null); + if (failedConnection != null) { + recovery = Mono.fromRunnable(() -> connectionCacheWrapper.invalidateConnection(failedConnection)) + .then(recovery); + } + } + + return recovery; + } + private String entityId() { return " " + ENTITY_PATH_KEY + ":" + entityName + (viaEntityName != null ? " " + VIA_ENTITY_NAME_KEY + ":" + viaEntityName : "") + " "; diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirer.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirer.java index ec8b8365ae7d1..b6b2c4f7883d6 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirer.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirer.java @@ -9,8 +9,10 @@ import com.azure.core.amqp.exception.AmqpException; import com.azure.core.amqp.implementation.StringUtil; import com.azure.core.amqp.implementation.handler.ReceiveLinkHandler2; +import com.azure.core.amqp.implementation.RecoveryKind; import com.azure.core.util.logging.ClientLogger; import com.azure.messaging.servicebus.implementation.MessagingEntityType; +import com.azure.messaging.servicebus.implementation.ServiceBusAmqpConnection; import com.azure.messaging.servicebus.implementation.ServiceBusManagementNode; import com.azure.messaging.servicebus.implementation.ServiceBusReceiveLink; import com.azure.messaging.servicebus.implementation.ServiceBusReceiveLink.SessionProperties; @@ -26,6 +28,7 @@ import java.time.OffsetDateTime; import java.util.Objects; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import static com.azure.core.amqp.implementation.ClientConstants.ENTITY_PATH_KEY; @@ -150,52 +153,80 @@ Mono acquire(String sessionId) { * is already disposed or {@link TimeoutException} if session acquire timeouts and {@code timeoutRetryDisabled} set to true. */ private Mono acquireIntern(String sessionId) { - if (timeoutRetryDisabled) { - return acquireSession(sessionId).timeout(clientSideTimeout, - Mono.error(() -> new TimeoutException("Timed out waiting to acquire a session; the broker did not " - + "respond within " + clientSideTimeout + " (client-side guard)."))) - .onErrorResume(t -> { - if (isBrokerTimeoutError(t)) { - // map the broker timeout to application-friendly TimeoutException. - final Throwable e = new TimeoutException("com.microsoft:timeout").initCause(t); - return publishError(sessionId, e, false); - } - if (t instanceof TimeoutException) { - // Client-side guard fired because the broker never responded (a hung acquire). - // Treat it like the broker timeout: a retriable "no session acquired in time". - return publishError(sessionId, t, false); - } - return publishError(sessionId, t, true); - }); - } else { - return acquireSession(sessionId).timeout(tryTimeout) - .retryWhen(Retry.from(signals -> signals.flatMap(signal -> { - final Throwable t = signal.failure(); - if (isTimeoutError(t)) { - logger.atVerbose() - .addKeyValue(ENTITY_PATH_KEY, entityPath) - .addKeyValue("attempt", signal.totalRetriesInARow()) - .log("Timeout while acquiring session '{}'.", sessionName(sessionId), t); - // Retry session acquire after a bounded backoff. Mono.delay hops to - // Schedulers.parallel(), freeing the QPid thread; the non-zero backoff prevents a - // tight CPU-burning loop when acquire attempts fail fast. - return Mono.delay(retryBackoff); - } - return publishError(sessionId, t, true); - }))); - } + return Mono.defer(() -> { + final AtomicReference operationConnection = new AtomicReference<>(); + if (timeoutRetryDisabled) { + return acquireSession(sessionId, operationConnection) + .timeout(clientSideTimeout, + Mono.error( + () -> new TimeoutException("Timed out waiting to acquire a session; the broker did not " + + "respond within " + clientSideTimeout + " (client-side guard)."))) + .onErrorResume(t -> { + if (isBrokerTimeoutError(t)) { + // map the broker timeout to application-friendly TimeoutException. + final Throwable e = new TimeoutException("com.microsoft:timeout").initCause(t); + return publishError(sessionId, e, false); + } + if (t instanceof TimeoutException) { + // Client-side guard fired because the broker never responded (a hung acquire). + // Treat it like the broker timeout: a retriable "no session acquired in time". + return publishError(sessionId, t, false); + } + return publishError(sessionId, t, true); + }); + } else { + return acquireSession(sessionId, operationConnection).timeout(tryTimeout) + .retryWhen(Retry.from(signals -> signals.flatMap(signal -> { + final Throwable t = signal.failure(); + final RecoveryKind kind = RecoveryKind.classify(t); + if (kind == RecoveryKind.CONNECTION) { + logger.atWarning() + .addKeyValue(ENTITY_PATH_KEY, entityPath) + .log("Connection-level error acquiring session, forcing connection recovery.", t); + final ServiceBusAmqpConnection failedConnection = operationConnection.getAndSet(null); + if (failedConnection != null) { + connectionCacheWrapper.invalidateConnection(failedConnection); + } + return Mono.delay(retryBackoff); + } + if (isTimeoutError(t)) { + logger.atVerbose() + .addKeyValue(ENTITY_PATH_KEY, entityPath) + .addKeyValue("attempt", signal.totalRetriesInARow()) + .log("Timeout while acquiring session '{}'.", sessionName(sessionId), t); + // Retry session acquire after a bounded backoff. Mono.delay hops to + // Schedulers.parallel(), freeing the QPid thread; the non-zero backoff prevents a + // tight CPU-burning loop when acquire attempts fail fast. + return Mono.delay(retryBackoff); + } + if (kind == RecoveryKind.LINK) { + logger.atWarning() + .addKeyValue(ENTITY_PATH_KEY, entityPath) + .log("Link-level error acquiring session, retrying.", t); + // Retry link-level failures after the same bounded backoff so repeated fast + // link detaches do not spin in a tight CPU-burning loop. + return Mono.delay(retryBackoff); + } + return publishError(sessionId, t, true); + }))); + } + }); } /** * Tries to acquire a session from the broker by opening an AMQP receive link. * * @param sessionId the unique id of the session to acquire, a value {@code null} means acquire any free session. + * @param operationConnection captures the connection used by the current acquisition attempt. * * @return the acquired session. */ - private Mono acquireSession(String sessionId) { + private Mono acquireSession(String sessionId, + AtomicReference operationConnection) { return Mono.defer(() -> { + operationConnection.set(null); final Mono createLink = connectionCacheWrapper.getConnection() + .doOnNext(operationConnection::set) .flatMap(connection -> connection.createReceiveLink(linkName(sessionId), entityPath, receiveMode, null, entityType, identifier, sessionId)); return createLink.flatMap(link -> { diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionManager.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionManager.java index b5760c5afe0ed..b3a9bff53ea65 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionManager.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionManager.java @@ -8,11 +8,13 @@ import com.azure.core.amqp.exception.AmqpException; import com.azure.core.amqp.exception.SessionErrorContext; import com.azure.core.amqp.implementation.MessageSerializer; +import com.azure.core.amqp.implementation.RecoveryKind; import com.azure.core.amqp.implementation.StringUtil; import com.azure.core.util.logging.ClientLogger; import com.azure.messaging.servicebus.implementation.DispositionStatus; import com.azure.messaging.servicebus.implementation.MessageUtils; import com.azure.messaging.servicebus.implementation.MessagingEntityType; +import com.azure.messaging.servicebus.implementation.ServiceBusAmqpConnection; import com.azure.messaging.servicebus.implementation.ServiceBusManagementNode; import com.azure.messaging.servicebus.implementation.ServiceBusReceiveLink; import com.azure.messaging.servicebus.implementation.instrumentation.ServiceBusTracer; @@ -36,6 +38,7 @@ import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -71,6 +74,9 @@ class ServiceBusSessionManager implements AutoCloseable, IServiceBusSessionManag private final Deque availableSchedulers = new ConcurrentLinkedDeque<>(); private final Duration maxSessionLockRenewDuration; private final Duration sessionIdleTimeout; + // Backoff between retries for link/connection-level failures, which can fail fast and would otherwise + // spin in a tight, CPU-burning loop. Honors the configured AmqpRetryOptions delay. + private final Duration retryBackoff; /** * SessionId to receiver mapping. @@ -116,6 +122,13 @@ class ServiceBusSessionManager implements AutoCloseable, IServiceBusSessionManag this.sessionIdleTimeout = receiverOptions.getSessionIdleTimeout() != null ? receiverOptions.getSessionIdleTimeout() : connectionCacheWrapper.getRetryOptions().getTryTimeout(); + // Link/connection-level acquire failures can fail fast (e.g. the broker repeatedly detaches the + // link quickly), so space those retries by the configured retry delay rather than retrying with a + // zero delay, which spins the retry loop. Falls back to 100ms when no usable delay is configured. + final Duration configuredDelay = connectionCacheWrapper.getRetryOptions().getDelay(); + this.retryBackoff = (configuredDelay == null || configuredDelay.isZero() || configuredDelay.isNegative()) + ? Duration.ofMillis(100) + : configuredDelay; } ServiceBusSessionManager(String entityPath, MessagingEntityType entityType, @@ -242,15 +255,20 @@ private AmqpErrorContext getErrorContext() { /** * Creates an session receive link. * + * @param operationConnection captures the connection used by the current acquisition attempt. * @return A Mono that completes with an session receive link. */ - private Mono createSessionReceiveLink() { + private Mono + createSessionReceiveLink(AtomicReference operationConnection) { final String sessionId = receiverOptions.getSessionId(); - final String linkName = (sessionId != null) ? sessionId : StringUtil.getRandomString("session-"); - return connectionCacheWrapper.getConnection().flatMap(connection -> { - return connection.createReceiveLink(linkName, entityPath, receiverOptions.getReceiveMode(), null, - entityType, identifier, sessionId); + return Mono.defer(() -> { + final String linkName = (sessionId != null) ? sessionId : StringUtil.getRandomString("session-"); + operationConnection.set(null); + return connectionCacheWrapper.getConnection().doOnNext(operationConnection::set).flatMap(connection -> { + return connection.createReceiveLink(linkName, entityPath, receiverOptions.getReceiveMode(), null, + entityType, identifier, sessionId); + }); }); } @@ -264,56 +282,80 @@ Mono getActiveLink() { if (this.receiveLink != null) { return Mono.just(this.receiveLink); } - return Mono.defer(() -> createSessionReceiveLink().flatMap(link -> link.getEndpointStates() - .filter(e -> e == AmqpEndpointState.ACTIVE) - .next() - // The reason for using 'switchIfEmpty' operator - - // - // While waiting for the link to ACTIVE, if the broker detaches the link without an error-condition, - // the link-endpoint-state publisher will transition to completion without ever emitting ACTIVE. Map - // such publisher completion to transient (i.e., retriable) AmqpException to enable retry. - // - // A detach without an error-condition can happen when Service upgrades. Also, while the service often - // detaches with the error-condition 'com.microsoft:timeout' when there is no session, sometimes, - // when a free or new session is unavailable, detach can happen without the error-condition. - // - .switchIfEmpty( - Mono.error(() -> new AmqpException(true, "Session receive link completed without being active", null))) - .timeout(operationTimeout) - .then(Mono.just(link)))).retryWhen(Retry.from(retrySignals -> retrySignals.flatMap(signal -> { - final Throwable failure = signal.failure(); - LOGGER.atInfo() + return Mono.defer(() -> { + final AtomicReference operationConnection = new AtomicReference<>(); + return createSessionReceiveLink(operationConnection).flatMap(link -> link.getEndpointStates() + .filter(e -> e == AmqpEndpointState.ACTIVE) + .next() + // The reason for using 'switchIfEmpty' operator - + // + // While waiting for the link to ACTIVE, if the broker detaches the link without an error-condition, + // the link-endpoint-state publisher will transition to completion without ever emitting ACTIVE. Map + // such publisher completion to transient (i.e., retriable) AmqpException to enable retry. + // + // A detach without an error-condition can happen when Service upgrades. Also, while the service often + // detaches with the error-condition 'com.microsoft:timeout' when there is no session, sometimes, + // when a free or new session is unavailable, detach can happen without the error-condition. + // + .switchIfEmpty(Mono + .error(() -> new AmqpException(true, "Session receive link completed without being active", null))) + .doOnCancel(() -> link.closeAsync().subscribe(unused -> { + }, error -> LOGGER.atWarning() .addKeyValue(ENTITY_PATH_KEY, entityPath) - .addKeyValue("attempt", signal.totalRetriesInARow()) - .log("Error occurred while getting unnamed session.", failure); - - if (isDisposed.get()) { - return Mono.error( - new AmqpException(false, "SessionManager is already disposed.", failure, getErrorContext())); - } else if (failure instanceof TimeoutException) { - return Mono.delay(Duration.ZERO); - } else if (failure instanceof AmqpException - && ((AmqpException) failure).getErrorCondition() == AmqpErrorCondition.TIMEOUT_ERROR) { - // The link closed remotely with 'Detach {errorCondition:com.microsoft:timeout}' frame because - // the broker waited for N seconds (60 sec hard limit today) but there was no free or new session. - // - // Given N seconds elapsed since the last session acquire attempt, request for a session on - // the 'parallel' Scheduler and free the 'QPid' thread for other IO. - // - return Mono.delay(Duration.ZERO); - } else { - final long id = System.nanoTime(); - LOGGER.atInfo().addKeyValue(TRACKING_ID_KEY, id).log("Unable to acquire new session.", failure); - // The link-endpoint-state publisher will emit signal on the reactor-executor thread, which is - // non-blocking, if we use the session processor to recover the error, it requires a blocking - // thread to close the client. Hence, we publish the error on the bounded-elastic thread. - return Mono.error(failure) - .publishOn(Schedulers.boundedElastic()) - .doOnError(e -> LOGGER.atInfo() - .addKeyValue(TRACKING_ID_KEY, id) - .log("Emitting the error signal received for session acquire attempt.", e)); - } - }))); + .log("Error closing abandoned session receive link.", error))) + .timeout(operationTimeout) + .then(Mono.just(link))).retryWhen(Retry.from(retrySignals -> retrySignals.flatMap(signal -> { + final Throwable failure = signal.failure(); + final RecoveryKind kind = RecoveryKind.classify(failure); + LOGGER.atInfo() + .addKeyValue(ENTITY_PATH_KEY, entityPath) + .addKeyValue("attempt", signal.totalRetriesInARow()) + .addKeyValue("recoveryKind", kind) + .log("Error occurred while getting unnamed session.", failure); + + if (isDisposed.get()) { + return Mono.error(new AmqpException(false, "SessionManager is already disposed.", failure, + getErrorContext())); + } + + if (kind == RecoveryKind.CONNECTION) { + LOGGER.atWarning() + .addKeyValue(ENTITY_PATH_KEY, entityPath) + .log("Connection-level error in session manager, forcing connection recovery.", failure); + final ServiceBusAmqpConnection failedConnection = operationConnection.getAndSet(null); + if (failedConnection != null) { + connectionCacheWrapper.invalidateConnection(failedConnection); + } + } + + // Mono.delay(Duration.ZERO) instead of Mono.empty() — the delay schedules the retry on the + // 'parallel' Scheduler, freeing the QPid (proton) thread for other IO. Using Mono.empty() would + // block the QPid thread and slow down / block message pumping in other sessions. + if (failure instanceof TimeoutException) { + return Mono.delay(Duration.ZERO); + } else if (failure instanceof AmqpException + && ((AmqpException) failure).getErrorCondition() == AmqpErrorCondition.TIMEOUT_ERROR) { + // The link closed remotely with 'Detach {errorCondition:com.microsoft:timeout}' frame because + // the broker waited for N seconds (60 sec hard limit today) but there was no free or new session. + return Mono.delay(Duration.ZERO); + } else if (kind == RecoveryKind.LINK || kind == RecoveryKind.CONNECTION) { + // Link or connection-level error — retry to acquire a fresh link (or connection) after a + // bounded backoff, so a link that repeatedly detaches quickly does not spin the retry loop. + return Mono.delay(retryBackoff); + } else { + final long id = System.nanoTime(); + LOGGER.atInfo().addKeyValue(TRACKING_ID_KEY, id).log("Unable to acquire new session.", failure); + // The link-endpoint-state publisher will emit signal on the reactor-executor thread, which is + // non-blocking, if we use the session processor to recover the error, it requires a blocking + // thread to close the client. Hence, we publish the error on the bounded-elastic thread. + return Mono.error(failure) + .publishOn(Schedulers.boundedElastic()) + .doOnError(e -> LOGGER.atInfo() + .addKeyValue(TRACKING_ID_KEY, id) + .log("Emitting the error signal received for session acquire attempt.", e)); + } + }))); + }); } /** diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorAmqpConnection.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorAmqpConnection.java index ecd6583f07a1c..b7c8ee2ce5575 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorAmqpConnection.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorAmqpConnection.java @@ -17,6 +17,7 @@ import com.azure.core.amqp.implementation.ReactorHandlerProvider; import com.azure.core.amqp.implementation.ReactorProvider; import com.azure.core.amqp.implementation.ReactorSession; +import com.azure.core.amqp.implementation.RecoveryKind; import com.azure.core.amqp.implementation.RequestResponseChannel; import com.azure.core.amqp.implementation.RequestResponseChannelCache; import com.azure.core.amqp.implementation.RetryUtil; @@ -205,8 +206,10 @@ public Mono createReceiveLink(String linkName, String ent LOGGER.atVerbose().addKeyValue(ENTITY_PATH_KEY, entityPath).log("Get or create consumer."); final AmqpRetryPolicy retryPolicy = RetryUtil.getRetryPolicy(retryOptions); - return session.createConsumer(linkName, entityPath, entityType, retryOptions.getTryTimeout(), retryPolicy, - receiveMode, clientIdentifier); + return session + .createConsumer(linkName, entityPath, entityType, retryOptions.getTryTimeout(), retryPolicy, + receiveMode, clientIdentifier) + .doOnError(error -> removeFailedSession(session, error)); }); } @@ -215,6 +218,11 @@ public Mono createSession(String sessionName) { return super.createSession(distributedTransactionsSupport ? CROSS_ENTITY_TRANSACTIONS_LINK_NAME : sessionName); } + @Override + public boolean removeSession(String sessionName) { + return super.removeSession(distributedTransactionsSupport ? CROSS_ENTITY_TRANSACTIONS_LINK_NAME : sessionName); + } + /** * Creates or gets an existing receive link. The same link is returned if there is an existing receive link with the * same {@code linkName}. Otherwise, a new link is created and returned. @@ -237,11 +245,28 @@ public Mono createReceiveLink(String linkName, String ent LOGGER.atVerbose().addKeyValue(ENTITY_PATH_KEY, entityPath).log("Get or create consumer."); final AmqpRetryPolicy retryPolicy = RetryUtil.getRetryPolicy(retryOptions); - return session.createConsumer(linkName, entityPath, entityType, retryOptions.getTryTimeout(), retryPolicy, - receiveMode, clientIdentifier, sessionId); + return session + .createConsumer(linkName, entityPath, entityType, retryOptions.getTryTimeout(), retryPolicy, + receiveMode, clientIdentifier, sessionId) + .doOnError(error -> removeFailedSession(session, error)); }); } + private void removeFailedSession(ServiceBusSession session, Throwable error) { + final RecoveryKind recoveryKind = RecoveryKind.classify(error); + if (recoveryKind != RecoveryKind.LINK && recoveryKind != RecoveryKind.CONNECTION) { + return; + } + try { + final boolean removed = removeSession(session.getSessionName(), session); + LOGGER.atVerbose() + .addKeyValue("sessionRemoved", removed) + .log("Stale session removal during {} recovery.", recoveryKind); + } catch (RuntimeException cleanupError) { + LOGGER.atWarning().log("Error removing stale session during {} recovery.", recoveryKind, cleanupError); + } + } + @Override protected ReactorSession createSession(ProtonSessionWrapper session) { return new ServiceBusReactorSession(this, session, handlerProvider, linkProvider, getClaimsBasedSecurityNode(), diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientTest.java index 8469c428b985e..b2f1a84176fb8 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientTest.java @@ -11,6 +11,7 @@ import com.azure.core.amqp.exception.AmqpErrorCondition; import com.azure.core.amqp.exception.AmqpException; import com.azure.core.amqp.implementation.ConnectionOptions; +import com.azure.core.amqp.implementation.ExceptionUtil; import com.azure.core.amqp.implementation.MessageSerializer; import com.azure.core.amqp.implementation.ReactorConnectionCache; import com.azure.core.amqp.models.CbsAuthorizationType; @@ -84,6 +85,7 @@ import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import java.util.stream.IntStream; import java.util.stream.Stream; @@ -1656,6 +1658,68 @@ private ServiceBusReceivedMessage mockReceivedMessage(Instant enqueuedTime) { return receivedMessage; } + @Test + void receiveResourceLimitRetriesWithoutRemovingTheSession() { + final Exception resourceLimit = ExceptionUtil + .toException(AmqpErrorCondition.RESOURCE_LIMIT_EXCEEDED.getErrorCondition(), "resource limit", null); + final AmqpException terminal = new AmqpException(false, AmqpErrorCondition.NOT_ALLOWED, "terminal", null); + final ConnectionCacheWrapper wrapper = mock(ConnectionCacheWrapper.class); + when(wrapper.getConnection()).thenReturn(Mono.just(connection)); + when(wrapper.getRetryOptions()).thenReturn(new AmqpRetryOptions().setDelay(Duration.ofMillis(1))); + when(wrapper.getFullyQualifiedNamespace()).thenReturn(NAMESPACE); + when(connection.createReceiveLink(anyString(), anyString(), any(ServiceBusReceiveMode.class), any(), + any(MessagingEntityType.class), anyString())).thenReturn(Mono.error(resourceLimit)) + .thenReturn(Mono.error(terminal)); + final ServiceBusReceiverAsyncClient client = new ServiceBusReceiverAsyncClient(NAMESPACE, ENTITY_PATH, + ENTITY_TYPE, createNonSessionOptions(ServiceBusReceiveMode.PEEK_LOCK, PREFETCH, null, false), wrapper, + CLEANUP_INTERVAL, instrumentation, messageSerializer, onClientClose, CLIENT_IDENTIFIER); + try { + StepVerifier.create(client.receiveMessages()) + .expectErrorSatisfies(error -> assertSame(terminal, error.getCause())) + .verify(DEFAULT_TIMEOUT); + verify(connection, times(2)).createReceiveLink(anyString(), anyString(), any(ServiceBusReceiveMode.class), + any(), any(MessagingEntityType.class), anyString()); + verify(connection, never()).removeSession(anyString()); + verify(wrapper, never()).invalidateConnection(any(ServiceBusAmqpConnection.class)); + } finally { + client.close(); + } + } + + @Test + void receiveRecoveryDoesNotAcquireOrRemoveTheReplacementSession() { + final ServiceBusAmqpConnection replacement = mock(ServiceBusAmqpConnection.class); + final AtomicReference current = new AtomicReference<>(connection); + final ConnectionCacheWrapper wrapper = mock(ConnectionCacheWrapper.class); + when(wrapper.getConnection()).thenReturn(Mono.fromSupplier(current::get)); + when(wrapper.getRetryOptions()).thenReturn(new AmqpRetryOptions().setDelay(Duration.ofMillis(1))); + when(wrapper.getFullyQualifiedNamespace()).thenReturn(NAMESPACE); + final AmqpException failed + = new AmqpException(true, AmqpErrorCondition.CONNECTION_FORCED, "connection forced", null); + final AmqpException terminal = new AmqpException(false, AmqpErrorCondition.NOT_ALLOWED, "terminal", null); + when(connection.createReceiveLink(anyString(), anyString(), any(ServiceBusReceiveMode.class), any(), + any(MessagingEntityType.class), anyString())).thenAnswer(invocation -> { + current.set(replacement); + return Mono.error(failed); + }); + when(replacement.createReceiveLink(anyString(), anyString(), any(ServiceBusReceiveMode.class), any(), + any(MessagingEntityType.class), anyString())).thenReturn(Mono.error(terminal)); + final ServiceBusReceiverAsyncClient client = new ServiceBusReceiverAsyncClient(NAMESPACE, ENTITY_PATH, + ENTITY_TYPE, createNonSessionOptions(ServiceBusReceiveMode.PEEK_LOCK, PREFETCH, null, false), wrapper, + CLEANUP_INTERVAL, instrumentation, messageSerializer, onClientClose, CLIENT_IDENTIFIER); + try { + StepVerifier.create(client.receiveMessages()) + .expectError(ServiceBusException.class) + .verify(DEFAULT_TIMEOUT); + verify(connection, never()).removeSession(anyString()); + verify(replacement, never()).removeSession(anyString()); + verify(wrapper).invalidateConnection(connection); + verify(wrapper, never()).invalidateConnection(replacement); + } finally { + client.close(); + } + } + private List getMessages() { final Map map = Collections.singletonMap("SAMPLE_HEADER", "foo"); diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientRecoveryIsolatedTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientRecoveryIsolatedTest.java index 9f70ed644f5e5..a6fa8c536bfa0 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientRecoveryIsolatedTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientRecoveryIsolatedTest.java @@ -6,12 +6,15 @@ import com.azure.core.amqp.AmqpEndpointState; import com.azure.core.amqp.AmqpRetryMode; import com.azure.core.amqp.AmqpRetryOptions; +import com.azure.core.amqp.AmqpSession; import com.azure.core.amqp.exception.AmqpErrorCondition; import com.azure.core.amqp.exception.AmqpErrorContext; import com.azure.core.amqp.exception.AmqpException; +import com.azure.core.amqp.exception.OperationCancelledException; import com.azure.core.amqp.implementation.AmqpLinkProvider; import com.azure.core.amqp.implementation.AmqpSendLink; import com.azure.core.amqp.implementation.ConnectionOptions; +import com.azure.core.amqp.implementation.ExceptionUtil; import com.azure.core.amqp.implementation.MessageSerializer; import com.azure.core.amqp.implementation.ReactorConnectionCache; import com.azure.core.amqp.implementation.ReactorDispatcher; @@ -20,6 +23,7 @@ import com.azure.core.amqp.implementation.ReactorProvider; import com.azure.core.amqp.implementation.TokenManager; import com.azure.core.amqp.implementation.TokenManagerProvider; +import com.azure.core.amqp.implementation.RetryUtil; import com.azure.core.amqp.implementation.handler.ConnectionHandler; import com.azure.core.amqp.implementation.handler.SendLinkHandler; import com.azure.core.amqp.implementation.handler.SessionHandler; @@ -27,7 +31,10 @@ import com.azure.messaging.servicebus.implementation.MessagingEntityType; import com.azure.messaging.servicebus.implementation.ServiceBusAmqpLinkProvider; import com.azure.messaging.servicebus.implementation.ServiceBusReactorAmqpConnection; +import com.azure.messaging.servicebus.implementation.instrumentation.ReceiverKind; +import com.azure.messaging.servicebus.implementation.instrumentation.ServiceBusReceiverInstrumentation; import com.azure.messaging.servicebus.implementation.instrumentation.ServiceBusSenderInstrumentation; +import com.azure.messaging.servicebus.models.ServiceBusReceiveMode; import org.apache.qpid.proton.amqp.transport.DeliveryState; import org.apache.qpid.proton.engine.Connection; import org.apache.qpid.proton.engine.EndpointState; @@ -40,6 +47,9 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.ValueSource; import org.junit.jupiter.api.parallel.Execution; import org.junit.jupiter.api.parallel.ExecutionMode; import org.junit.jupiter.api.parallel.Isolated; @@ -67,6 +77,7 @@ import java.util.List; import java.util.Objects; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -654,10 +665,227 @@ private MockEndpoint createMockEndpoint(int sessionsCnt, int[] linksPerSession) return MockEndpoint.create(connectionId, QUEUE_NAME, RETRY_OPTIONS, sessionsCnt, linksPerSession); } + @Test + void resourceLimitFromExceptionUtilRetriesWithoutClosingResources() { + final Exception error + = ExceptionUtil.toException(AmqpErrorCondition.RESOURCE_LIMIT_EXCEEDED.getErrorCondition(), + "resource limit", new AmqpErrorContext(FQDN)); + Assertions.assertFalse(((AmqpException) error).isTransient()); + final AmqpException retriable = (AmqpException) RecoveryUtils.asRetriable(error); + Assertions.assertTrue(retriable.isTransient()); + Assertions.assertEquals(AmqpErrorCondition.RESOURCE_LIMIT_EXCEEDED, retriable.getErrorCondition()); + Assertions.assertSame(((AmqpException) error).getContext(), retriable.getContext()); + Assertions.assertSame(error, retriable.getCause()); + try (MockEndpoint endpoint = createMockEndpoint(1, new int[] { 1 })) { + final AmqpSendLink link = endpoint.getAmqpSendLink(0, 0); + when(link.send(any(Message.class))).thenReturn(Mono.error(error)).thenReturn(Mono.empty()); + final ReactorConnectionCache cache + = createConnectionCache(singleConnectionSupplier(endpoint)); + final ServiceBusSenderAsyncClient client = createSenderAsyncClient(cache, false); + try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) { + verifier.create(() -> client.sendMessage(createMessageToSend())) + .thenAwait(VIRTUAL_TIME_SHIFT) + .expectComplete() + .verify(Duration.ofSeconds(5)); + verify(link, times(2)).send(any(Message.class)); + verify(link, Mockito.never()).closeAsync(); + } finally { + client.close(); + cache.dispose(); + } + } + } + + @ParameterizedTest + @EnumSource( + value = AmqpErrorCondition.class, + names = { "NOT_FOUND", "UNAUTHORIZED_ACCESS", "MESSAGE_LOCK_LOST", "SESSION_CANNOT_BE_LOCKED" }) + void nonRetriableErrorsStayNonRetriable(AmqpErrorCondition condition) { + final AmqpException error = new AmqpException(false, condition, "terminal", null); + Assertions.assertSame(error, RecoveryUtils.asRetriable(error)); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void sessionRemovalUsesTheSameKeyAsCreation(boolean crossEntityTransactions) { + try (MockEndpoint endpoint = createMockEndpoint(2, new int[] { 0, 0 })) { + final ReactorConnectionCache cache + = createConnectionCache(() -> endpoint.arrange(crossEntityTransactions)); + try { + final ServiceBusReactorAmqpConnection connection = cache.get().block(Duration.ofSeconds(5)); + final AmqpSession first = connection.createSession(QUEUE_NAME).block(Duration.ofSeconds(5)); + Assertions.assertNotNull(first); + Assertions.assertTrue(connection.removeSession(QUEUE_NAME)); + final AmqpSession replacement = connection.createSession(QUEUE_NAME).block(Duration.ofSeconds(5)); + Assertions.assertNotNull(replacement); + Assertions.assertNotSame(first, replacement); + } finally { + cache.dispose(); + } + } + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void delayedReceiveLinkFailureDoesNotRemoveReplacementSession(boolean crossEntityTransactions) { + try (MockEndpoint endpoint = createMockEndpoint(2, new int[] { 0, 0 })) { + final ReactorConnectionCache cache + = createConnectionCache(() -> Mockito.spy(endpoint.arrange(crossEntityTransactions))); + final ServiceBusReactorAmqpConnection connection = cache.get().block(Duration.ofSeconds(5)); + final AmqpSession failedSession = connection.createSession(QUEUE_NAME).block(Duration.ofSeconds(5)); + final AtomicReference replacement = new AtomicReference<>(); + final Sinks.One failureHandled = Sinks.one(); + Mockito.doAnswer(invocation -> { + Assertions.assertTrue(connection.removeSession(QUEUE_NAME)); + replacement.set(connection.createSession(QUEUE_NAME).block(Duration.ofSeconds(5))); + Assertions.assertNotSame(failedSession, replacement.get()); + return Mono.error(RETRIABLE_LINK_ERROR) + .doAfterTerminate(() -> failureHandled.emitEmpty(Sinks.EmitFailureHandler.FAIL_FAST)); + }) + .doReturn(Mono.error(NON_RETRIABLE_ERROR_1)) + .when(connection) + .createReceiveLink(anyString(), anyString(), any(ServiceBusReceiveMode.class), any(), + any(MessagingEntityType.class), anyString()); + final ServiceBusReceiverAsyncClient client = new ServiceBusReceiverAsyncClient(FQDN, QUEUE_NAME, + MessagingEntityType.QUEUE, + ReceiverOptions.createNonSessionOptions(ServiceBusReceiveMode.PEEK_LOCK, 1, null, false), + new ConnectionCacheWrapper(cache), Duration.ofSeconds(10), + new ServiceBusReceiverInstrumentation(null, null, FQDN, QUEUE_NAME, null, ReceiverKind.ASYNC_RECEIVER), + messageSerializer, onClientClosed, CLIENT_IDENTIFIER); + try { + StepVerifier.create(client.receiveMessages()).then(() -> { + failureHandled.asMono().block(Duration.ofSeconds(5)); + Assertions.assertNotNull(replacement.get()); + Assertions.assertFalse(replacement.get().isDisposed()); + }).thenCancel().verify(Duration.ofSeconds(5)); + Assertions.assertSame(connection, cache.get().block(Duration.ofSeconds(5))); + } finally { + client.close(); + cache.dispose(); + } + } + } + + @Test + void shouldRecoverAfterNegotiationRetriesAreExhausted() { + try (MockEndpoint endpoint = createMockEndpoint(1, new int[] { 2 })) { + final AmqpSendLink failed = endpoint.getAmqpSendLink(0, 0); + final AmqpRetryOptions negotiationOptions = new AmqpRetryOptions(RETRY_OPTIONS).setMaxRetries(0); + when(failed.closeAsync()).thenAnswer(invocation -> { + endpoint.emitCurrentSendLinkError(RETRIABLE_LINK_ERROR); + return Mono.empty(); + }); + when(endpoint.getAmqpSendLink(0, 1).send(any(Message.class))).thenReturn(Mono.empty()); + final ReactorConnectionCache cache = createConnectionCache(() -> { + final ServiceBusReactorAmqpConnection result = endpoint.arrange(); + when(failed.getLinkSize()).thenReturn( + RetryUtil.withRetry(Mono.error(RETRIABLE_LINK_ERROR), negotiationOptions, "negotiation")); + return result; + }); + final ServiceBusSenderAsyncClient client = createSenderAsyncClient(cache, false); + try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) { + verifier.create(() -> client.sendMessage(createMessageToSend())) + .thenAwait(VIRTUAL_TIME_SHIFT) + .expectComplete() + .verify(Duration.ofSeconds(5)); + } finally { + client.close(); + cache.dispose(); + } + } + } + + @Test + void shouldInvalidateConnectionBeforeNonCompletingLinkClose() { + final List counts = new ArrayList<>(); + counts.add(new SessionLinkCount(1, new int[] { 1 })); + counts.add(new SessionLinkCount(1, new int[] { 1 })); + try (MockEndpoints endpoints = createMockEndpoints(counts)) { + final AtomicInteger supplied = new AtomicInteger(); + final ReactorConnectionCache cache + = createConnectionCache(() -> endpoints.get(supplied.getAndIncrement()).arrange()); + final AmqpSendLink failed = endpoints.get(0).getAmqpSendLink(0, 0); + when(failed.send(any(Message.class))).thenReturn(Mono.error(RETRIABLE_CONNECTION_ERROR)); + when(failed.closeAsync()).thenReturn(Mono.never()); + when(endpoints.get(1).getAmqpSendLink(0, 0).send(any(Message.class))).thenReturn(Mono.empty()); + final ServiceBusSenderAsyncClient client = createSenderAsyncClient(cache, false); + try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) { + verifier.create(() -> client.sendMessage(createMessageToSend())) + .thenAwait(VIRTUAL_TIME_SHIFT) + .expectComplete() + .verify(Duration.ofSeconds(5)); + Assertions.assertEquals(2, supplied.get()); + } finally { + client.close(); + cache.dispose(); + } + } + } + + @Test + void delayedConnectionErrorMustNotInvalidateReplacement() { + final List counts = new ArrayList<>(); + counts.add(new SessionLinkCount(1, new int[] { 1 })); + counts.add(new SessionLinkCount(1, new int[] { 1 })); + try (MockEndpoints endpoints = createMockEndpoints(counts)) { + final AtomicInteger supplied = new AtomicInteger(); + final ReactorConnectionCache cache + = createConnectionCache(() -> endpoints.get(supplied.getAndIncrement()).arrange()); + final AmqpSendLink failed = endpoints.get(0).getAmqpSendLink(0, 0); + when(failed.send(any(Message.class))).thenAnswer(invocation -> { + endpoints.get(0).emitConnectionError(RETRIABLE_CONNECTION_ERROR); + return cache.get().then(Mono.error(RETRIABLE_CONNECTION_ERROR)); + }); + when(failed.closeAsync()).thenReturn(Mono.empty()); + when(endpoints.get(1).getAmqpSendLink(0, 0).send(any(Message.class))).thenReturn(Mono.empty()); + final ServiceBusSenderAsyncClient client = createSenderAsyncClient(cache, false); + try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) { + verifier.create(() -> client.sendMessage(createMessageToSend())) + .thenAwait(VIRTUAL_TIME_SHIFT) + .expectComplete() + .verify(Duration.ofSeconds(5)); + Assertions.assertEquals(2, supplied.get()); + } finally { + client.close(); + cache.dispose(); + } + } + } + private MockEndpoints createMockEndpoints(List sessionLinkCountList) { return MockEndpoints.create(QUEUE_NAME, RETRY_OPTIONS, sessionLinkCountList); } + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void abortedSendRecoversItsLink(boolean withCause) { + try (MockEndpoint endpoint = createMockEndpoint(1, new int[] { 2 })) { + final OperationCancelledException aborted = withCause + ? new OperationCancelledException("Delivery advance failed", new IllegalStateException("advance"), null) + : new OperationCancelledException("Released", null); + final AmqpSendLink failed = endpoint.getAmqpSendLink(0, 0); + when(failed.send(any(Message.class))).thenReturn(Mono.error(aborted)); + when(failed.closeAsync()).thenAnswer(invocation -> { + endpoint.emitCurrentSendLinkCompletion(); + return Mono.empty(); + }); + when(endpoint.getAmqpSendLink(0, 1).send(any(Message.class))).thenReturn(Mono.empty()); + final ReactorConnectionCache cache + = createConnectionCache(singleConnectionSupplier(endpoint)); + final ServiceBusSenderAsyncClient client = createSenderAsyncClient(cache, false); + try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) { + verifier.create(() -> client.sendMessage(createMessageToSend())) + .thenAwait(VIRTUAL_TIME_SHIFT) + .expectComplete() + .verify(Duration.ofSeconds(5)); + verify(failed).closeAsync(); + } finally { + client.close(); + cache.dispose(); + } + } + } + private Supplier singleConnectionSupplier(MockEndpoint endpoint) { final Supplier connectionSupplier = new Supplier() { @@ -849,6 +1077,10 @@ static MockEndpoint create(String connectionId, String queueName, AmqpRetryOptio } ServiceBusReactorAmqpConnection arrange() { + return arrange(false); + } + + ServiceBusReactorAmqpConnection arrange(boolean crossEntityTransactions) { if (arranged.getAndSet(true)) { throw new RuntimeException("Only one connection can be obtained from a MockEndpoint instance."); } @@ -889,7 +1121,8 @@ ServiceBusReactorAmqpConnection arrange() { // New tests only for ReactorConnectionCache introduced in v2. final boolean isV2 = true; return new ServiceBusReactorAmqpConnection(connectionId, connectionOptions, reactorProvider, - handlerProvider, linkProvider, tokenManagerProvider, messageSerializer, false, isV2, false); + handlerProvider, linkProvider, tokenManagerProvider, messageSerializer, crossEntityTransactions, isV2, + false); } AmqpSendLink getAmqpSendLink(int sessionIdx, int linkIdx) { diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientTest.java index e80bdd93f03cc..c2a7c553b7767 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientTest.java @@ -6,6 +6,8 @@ import com.azure.core.amqp.AmqpEndpointState; import com.azure.core.amqp.AmqpRetryMode; import com.azure.core.amqp.AmqpRetryOptions; +import com.azure.core.amqp.exception.AmqpErrorCondition; +import com.azure.core.amqp.exception.AmqpException; import com.azure.core.amqp.AmqpTransaction; import com.azure.core.amqp.AmqpTransportType; import com.azure.core.amqp.FixedAmqpRetryPolicy; @@ -778,7 +780,8 @@ void failedSendMessageReportsMetrics(boolean isV2) { when(connection.createSendLink(eq(ENTITY_NAME), eq(ENTITY_NAME), eq(retryOptions), isNull(), eq(CLIENT_IDENTIFIER))).thenReturn(Mono.just(sendLink)); - when(sendLink.send(any(Message.class))).thenThrow(new RuntimeException("foo")); + when(sendLink.send(any(Message.class))) + .thenReturn(Mono.error(new AmqpException(false, AmqpErrorCondition.NOT_FOUND, "entity not found", null))); // Act StepVerifier.create(sender.sendMessage(new ServiceBusMessage(TEST_CONTENTS))) diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirerIsolatedTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirerIsolatedTest.java index 321acb70c2662..a95f98010c76b 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirerIsolatedTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionAcquirerIsolatedTest.java @@ -32,6 +32,7 @@ import java.util.ArrayDeque; import java.util.Deque; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import static org.mockito.ArgumentMatchers.any; @@ -159,6 +160,32 @@ public void shouldBackOffBetweenRetriesIfRetryEnabled() { Assertions.assertEquals(0, onCreateSessionLink.pending()); } + @Test + @Execution(ExecutionMode.SAME_THREAD) + public void shouldBackOffBetweenConnectionRecoveryAttempts() { + final AmqpException connectionError + = new AmqpException(true, AmqpErrorCondition.CONNECTION_FORCED, "connection forced", null); + final RuntimeException terminal = new RuntimeException("stop after two failures"); + final Deque> links = new ArrayDeque<>(); + links.add(Mono.error(connectionError)); + links.add(Mono.error(connectionError)); + links.add(Mono.error(terminal)); + final OnCreateSessionLink onCreate = new OnCreateSessionLink(links); + final ConnectionCacheWrapper wrapper = createMockConnectionWrapper(onCreate); + final ServiceBusSessionAcquirer acquirer = createSessionAcquirer(wrapper, false); + + try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) { + verifier.create(acquirer::acquire) + .thenAwait(RETRY_BACKOFF.dividedBy(2)) + .then(() -> Assertions.assertEquals(2, onCreate.pending())) + .thenAwait(RETRY_BACKOFF) + .then(() -> Assertions.assertEquals(1, onCreate.pending())) + .thenAwait(RETRY_BACKOFF) + .verifyErrorSatisfies(error -> Assertions.assertSame(terminal, error)); + } + Mockito.verify(wrapper, Mockito.times(2)).invalidateConnection(any(ServiceBusAmqpConnection.class)); + } + @Test @Execution(ExecutionMode.SAME_THREAD) public void shouldRetryOnBrokerTimeoutErrorIfRetryEnabled() { @@ -211,12 +238,60 @@ private ConnectionCacheWrapper createMockConnectionWrapper(OnCreateSessionLink o return cacheWrapper; } + @Test + public void shouldPropagateLockedNamedSessionWithoutRetry() { + final AmqpException locked = new AmqpException(false, AmqpErrorCondition.SESSION_CANNOT_BE_LOCKED, + "The requested session is already locked.", new AmqpErrorContext(ENTITY_PATH)); + final Deque> sessionLinks = new ArrayDeque<>(); + sessionLinks.add(Mono.error(locked)); + sessionLinks.add(Mono.error(new IllegalStateException("Unexpected second acquisition"))); + final OnCreateSessionLink onCreateSessionLink = new OnCreateSessionLink(sessionLinks); + final ServiceBusSessionAcquirer acquirer + = createSessionAcquirer(createMockConnectionWrapper(onCreateSessionLink), false); + + try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) { + verifier.create(() -> acquirer.acquire("locked-session")) + .thenAwait(RETRY_BACKOFF.plusSeconds(1)) + .verifyErrorSatisfies(error -> Assertions.assertSame(locked, error)); + } + Assertions.assertEquals(1, onCreateSessionLink.pending()); + } + private ServiceBusSessionAcquirer createSessionAcquirer(ConnectionCacheWrapper cacheWrapper, boolean isTimeoutRetryDisabled) { return new ServiceBusSessionAcquirer(LOGGER, IDENTIFIER, ENTITY_PATH, ENTITY_TYPE, RECEIVE_MODE, TRY_TIMEOUT, isTimeoutRetryDisabled, cacheWrapper); } + @Test + public void connectionRecoveryTargetsTheFailedAcquisition() { + final ServiceBusAmqpConnection failed = mock(ServiceBusAmqpConnection.class); + final ServiceBusAmqpConnection replacement = mock(ServiceBusAmqpConnection.class); + final AtomicReference current = new AtomicReference<>(failed); + final ConnectionCacheWrapper wrapper = mock(ConnectionCacheWrapper.class); + when(wrapper.isV2()).thenReturn(true); + when(wrapper.getConnection()).thenReturn(Mono.fromSupplier(current::get)); + when(wrapper.getRetryOptions()).thenReturn(new AmqpRetryOptions().setDelay(RETRY_BACKOFF)); + final AmqpException connectionError + = new AmqpException(true, AmqpErrorCondition.CONNECTION_FORCED, "connection forced", null); + final IllegalStateException terminal = new IllegalStateException("stop after replacement"); + when(failed.createReceiveLink(anyString(), anyString(), any(ServiceBusReceiveMode.class), any(), + any(MessagingEntityType.class), anyString(), any())).thenAnswer(invocation -> { + current.set(replacement); + return Mono.error(connectionError); + }); + when(replacement.createReceiveLink(anyString(), anyString(), any(ServiceBusReceiveMode.class), any(), + any(MessagingEntityType.class), anyString(), any())).thenReturn(Mono.error(terminal)); + + try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) { + verifier.create(() -> createSessionAcquirer(wrapper, false).acquire("session")) + .thenAwait(AWAIT_DURATION) + .verifyErrorSatisfies(error -> Assertions.assertSame(terminal, error)); + } + Mockito.verify(wrapper).invalidateConnection(failed); + Mockito.verify(wrapper, Mockito.never()).invalidateConnection(replacement); + } + private static final class OnCreateSessionLink implements Answer> { final Deque> sessionLinks = new ArrayDeque<>(); diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionManagerTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionManagerTest.java index 8347fe3520f7a..a7c7d1d24b90c 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionManagerTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionManagerTest.java @@ -7,6 +7,8 @@ import com.azure.core.amqp.AmqpRetryOptions; import com.azure.core.amqp.AmqpTransportType; import com.azure.core.amqp.ProxyOptions; +import com.azure.core.amqp.exception.AmqpErrorCondition; +import com.azure.core.amqp.exception.AmqpException; import com.azure.core.amqp.implementation.ConnectionOptions; import com.azure.core.amqp.implementation.MessageSerializer; import com.azure.core.amqp.models.CbsAuthorizationType; @@ -51,12 +53,14 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import static com.azure.messaging.servicebus.ReceiverOptions.createUnnamedSessionOptions; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; @@ -172,6 +176,81 @@ void properties() { assertEquals(CLIENT_IDENTIFIER, sessionManager.getIdentifier()); } + @Test + void lockedNamedSessionIsTerminal() { + final String sessionId = "locked-session"; + final AmqpException locked = new AmqpException(false, AmqpErrorCondition.SESSION_CANNOT_BE_LOCKED, + "The requested session is already locked.", null); + final ReceiverOptions options = ReceiverOptions.createNamedSessionOptions(ServiceBusReceiveMode.PEEK_LOCK, 1, + MAX_LOCK_RENEWAL, false, sessionId); + sessionManager = new ServiceBusSessionManager(ENTITY_PATH, ENTITY_TYPE, connectionCacheWrapper, + messageSerializer, options, CLIENT_IDENTIFIER, NOOP_TRACER); + when(connection.createReceiveLink(eq(sessionId), eq(ENTITY_PATH), any(ServiceBusReceiveMode.class), isNull(), + eq(ENTITY_TYPE), eq(CLIENT_IDENTIFIER), eq(sessionId))).thenReturn(Mono.error(locked)); + + StepVerifier.create(sessionManager.getActiveLink()) + .expectErrorSatisfies(error -> assertSame(locked, error)) + .verify(Duration.ofSeconds(2)); + verify(connection, times(1)).createReceiveLink(eq(sessionId), eq(ENTITY_PATH), any(ServiceBusReceiveMode.class), + isNull(), eq(ENTITY_TYPE), eq(CLIENT_IDENTIFIER), eq(sessionId)); + } + + @Test + void unnamedSessionRetriesWithFreshLinkNameAfterTimeout() { + connectionCacheWrapper.getRetryOptions().setTryTimeout(Duration.ofMillis(100)); + final ReceiverOptions options = createUnnamedSessionOptions(ServiceBusReceiveMode.PEEK_LOCK, 1, + MAX_LOCK_RENEWAL, false, 5, SESSION_IDLE_TIMEOUT); + sessionManager = new ServiceBusSessionManager(ENTITY_PATH, ENTITY_TYPE, connectionCacheWrapper, + messageSerializer, options, CLIENT_IDENTIFIER, NOOP_TRACER); + final ServiceBusReceiveLink hungLink = mock(ServiceBusReceiveLink.class); + when(hungLink.getEndpointStates()).thenReturn(Flux.never()); + final AtomicBoolean abandonedLinkClosed = new AtomicBoolean(); + final Sinks.Empty closeCompletion = Sinks.empty(); + when(hungLink.closeAsync()) + .thenReturn(closeCompletion.asMono().doOnSubscribe(ignored -> abandonedLinkClosed.set(true))); + final AtomicReference firstLinkName = new AtomicReference<>(); + when(connection.createReceiveLink(anyString(), eq(ENTITY_PATH), any(ServiceBusReceiveMode.class), isNull(), + eq(ENTITY_TYPE), eq(CLIENT_IDENTIFIER), isNull())).thenAnswer(invocation -> { + final String linkName = invocation.getArgument(0); + firstLinkName.compareAndSet(null, linkName); + return Mono.just(firstLinkName.get().equals(linkName) ? hungLink : amqpReceiveLink); + }); + + StepVerifier.create(sessionManager.getActiveLink()) + .assertNext(link -> assertSame(amqpReceiveLink, link)) + .expectComplete() + .verify(Duration.ofSeconds(2)); + verify(connection, times(2)).createReceiveLink(linkNameCaptor.capture(), eq(ENTITY_PATH), + any(ServiceBusReceiveMode.class), isNull(), eq(ENTITY_TYPE), eq(CLIENT_IDENTIFIER), isNull()); + assertEquals(2, linkNameCaptor.getAllValues().stream().distinct().count()); + assertTrue(abandonedLinkClosed.get(), "the abandoned link must be closed independently of the retry"); + verify(amqpReceiveLink, Mockito.never()).closeAsync(); + closeCompletion.tryEmitEmpty(); + } + + @Test + void cancellingUnnamedSessionAcquisitionClosesOnlyThePendingLink() { + final ReceiverOptions options = createUnnamedSessionOptions(ServiceBusReceiveMode.PEEK_LOCK, 1, + MAX_LOCK_RENEWAL, false, 5, SESSION_IDLE_TIMEOUT); + sessionManager = new ServiceBusSessionManager(ENTITY_PATH, ENTITY_TYPE, connectionCacheWrapper, + messageSerializer, options, CLIENT_IDENTIFIER, NOOP_TRACER); + final ServiceBusReceiveLink pending = mock(ServiceBusReceiveLink.class); + when(pending.getEndpointStates()).thenReturn(Flux.never()); + final AtomicBoolean closed = new AtomicBoolean(); + final Sinks.Empty closeCompletion = Sinks.empty(); + when(pending.closeAsync()).thenReturn(closeCompletion.asMono().doOnSubscribe(ignored -> closed.set(true))); + when(connection.createReceiveLink(anyString(), eq(ENTITY_PATH), any(ServiceBusReceiveMode.class), isNull(), + eq(ENTITY_TYPE), eq(CLIENT_IDENTIFIER), isNull())).thenReturn(Mono.just(pending)); + + StepVerifier.create(sessionManager.getActiveLink()) + .thenAwait(Duration.ofMillis(10)) + .thenCancel() + .verify(Duration.ofSeconds(2)); + assertTrue(closed.get(), "cancellation must close the link before it can acquire an unconsumed session"); + verify(pending).closeAsync(); + closeCompletion.tryEmitEmpty(); + } + @Test void receiveNull() { // Arrange diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorAmqpConnectionTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorAmqpConnectionTest.java new file mode 100644 index 0000000000000..f86752e57e4cc --- /dev/null +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorAmqpConnectionTest.java @@ -0,0 +1,111 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.servicebus.implementation; + +import com.azure.core.amqp.AmqpRetryOptions; +import com.azure.core.amqp.AmqpSession; +import com.azure.core.amqp.exception.AmqpErrorCondition; +import com.azure.core.amqp.exception.AmqpException; +import com.azure.core.amqp.implementation.ConnectionOptions; +import com.azure.core.amqp.implementation.ReactorHandlerProvider; +import com.azure.core.amqp.implementation.ReactorProvider; +import com.azure.core.amqp.implementation.TokenManagerProvider; +import com.azure.core.amqp.implementation.handler.ConnectionHandler; +import com.azure.messaging.servicebus.models.ServiceBusReceiveMode; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.mockito.Mockito; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; +import reactor.test.StepVerifier; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class ServiceBusReactorAmqpConnectionTest { + @AfterEach + void teardown() { + Mockito.framework().clearInlineMock(this); + } + + @ParameterizedTest + @CsvSource({ + "false,false,LINK_DETACH_FORCED", + "false,true,LINK_DETACH_FORCED", + "true,false,LINK_DETACH_FORCED", + "true,true,LINK_DETACH_FORCED", + "false,false,CONNECTION_FORCED", + "false,true,CONNECTION_FORCED", + "true,false,CONNECTION_FORCED", + "true,true,CONNECTION_FORCED", + "false,false,RESOURCE_LIMIT_EXCEEDED", + "false,true,RESOURCE_LIMIT_EXCEEDED", + "true,false,RESOURCE_LIMIT_EXCEEDED", + "true,true,RESOURCE_LIMIT_EXCEEDED", + "false,false,NOT_ALLOWED", + "false,true,NOT_ALLOWED", + "true,false,NOT_ALLOWED", + "true,true,NOT_ALLOWED" }) + void receiveLinkFailureRemovesOnlyItsSession(boolean sessionReceiver, boolean replaceBeforeFailure, + AmqpErrorCondition condition) { + final ServiceBusSession failed = mock(ServiceBusSession.class); + final ServiceBusSession replacement = mock(ServiceBusSession.class); + when(failed.getSessionName()).thenReturn("cached-session"); + final Sinks.One pending = Sinks.one(); + when(failed.createConsumer(anyString(), anyString(), any(), any(), any(), any(), anyString())) + .thenReturn(pending.asMono()); + when(failed.createConsumer(anyString(), anyString(), any(), any(), any(), any(), anyString(), anyString())) + .thenReturn(pending.asMono()); + final AtomicReference cached = new AtomicReference<>(failed); + final ConnectionOptions options = mock(ConnectionOptions.class); + when(options.getRetry()).thenReturn(new AmqpRetryOptions()); + final ReactorHandlerProvider handlers = mock(ReactorHandlerProvider.class); + final ConnectionHandler handler = mock(ConnectionHandler.class); + when(handlers.createConnectionHandler(anyString(), any())).thenReturn(handler); + when(handler.getEndpointStates()).thenReturn(Flux.never()); + final ServiceBusReactorAmqpConnection connection + = new ServiceBusReactorAmqpConnection("connection", options, mock(ReactorProvider.class), handlers, + mock(ServiceBusAmqpLinkProvider.class), mock(TokenManagerProvider.class), null, false, true, false) { + @Override + public Mono createSession(String name) { + return Mono.just(cached.get()); + } + + @Override + protected boolean removeSession(String name, AmqpSession expected) { + Assertions.assertEquals("cached-session", name); + Assertions.assertSame(failed, expected); + return cached.compareAndSet(expected, null); + } + }; + final AmqpException error = new AmqpException(true, condition, "receive failure", null); + final Mono link = sessionReceiver + ? connection.createReceiveLink("link", "entity", ServiceBusReceiveMode.PEEK_LOCK, null, + MessagingEntityType.QUEUE, "client", "session-id") + : connection.createReceiveLink("link", "entity", ServiceBusReceiveMode.PEEK_LOCK, null, + MessagingEntityType.QUEUE, "client"); + StepVerifier.create(link).then(() -> { + if (replaceBeforeFailure) { + cached.set(replacement); + } + pending.emitError(error, Sinks.EmitFailureHandler.FAIL_FAST); + }).expectErrorMatches(actual -> actual == error).verify(Duration.ofSeconds(5)); + if (replaceBeforeFailure) { + Assertions.assertSame(replacement, cached.get()); + } else if (condition == AmqpErrorCondition.LINK_DETACH_FORCED + || condition == AmqpErrorCondition.CONNECTION_FORCED) { + Assertions.assertNull(cached.get()); + } else { + Assertions.assertSame(failed, cached.get()); + } + } +} diff --git a/sdk/servicebus/azure-resourcemanager-servicebus/pom.xml b/sdk/servicebus/azure-resourcemanager-servicebus/pom.xml index 373c0bf21edaa..820f6ebabf329 100644 --- a/sdk/servicebus/azure-resourcemanager-servicebus/pom.xml +++ b/sdk/servicebus/azure-resourcemanager-servicebus/pom.xml @@ -55,6 +55,11 @@ + + com.azure + azure-core + 1.61.0 + com.azure.resourcemanager azure-resourcemanager-resources diff --git a/sdk/servicebus/ci.yml b/sdk/servicebus/ci.yml index 58823bab7a695..49564ad549aa5 100644 --- a/sdk/servicebus/ci.yml +++ b/sdk/servicebus/ci.yml @@ -13,6 +13,7 @@ trigger: - sdk/servicebus/azure-messaging-servicebus-stress/ - sdk/servicebus/azure-messaging-servicebus-track2-perf/ - sdk/servicebus/build/ + - sdk/core/azure-core-amqp/ exclude: - sdk/servicebus/pom.xml - sdk/servicebus/azure-messaging-servicebus/pom.xml @@ -33,6 +34,7 @@ pr: - sdk/servicebus/azure-messaging-servicebus-stress/ - sdk/servicebus/azure-messaging-servicebus-track2-perf/ - sdk/servicebus/build/ + - sdk/core/azure-core-amqp/ exclude: - sdk/servicebus/pom.xml - sdk/servicebus/azure-messaging-servicebus/pom.xml @@ -59,3 +61,6 @@ extends: # required by the above perf libraries - name: perf-test-core groupId: com.azure + - name: azure-core-amqp + groupId: com.azure + # Build azure-core-amqp from source (needed for RecoveryKind, tiered retry)