[BUG] - Handler.close livelocks on FAIL_NON_SERIALIZED while holding the ReactorConnection monitor

[BUG] Handler.close livelocks on FAIL_NON_SERIALIZED while holding the ReactorConnection monitor

Describe the bug

Handler.close() retries endpointStates.emitNext(CLOSED) forever when the result is FAIL_NON_SERIALIZED. ReactorConnection.closeConnectionWork() calls close() while holding the connection monitor (private synchronized). If reactor-executor is already inside emitNext on that same sink, the retry never succeeds, the monitor is never released, and the reactor thread blocks forever in getOrCreateConnection().

The loop uses about one core and allocates on the order of 1 GiB/s of log events. It does not time out. In the occurrence below it ran until the process was restarted, more than a day later. Message settlement on that connection stopped for the whole interval.

This is not #39999. That report is the Event Hubs buffered producer logging FAIL_NON_SERIALIZED, retrying a bounded number of times, and then failing the send. Handler.close() returns true with no limit and no backoff, and it does so while holding the connection monitor.

Exception or Stack Trace

No exception escapes. The close path logs this once per retry:

WebSocketsConnectionHandler {"az.sdk.message":"Could not emit closed endpoint state. Non-serial access. Retrying.","connectionId":"MF_b63740_...","signalType":"onNext","emitResult":"FAIL_NON_SERIALIZED"}

About 3.4e7 of those lines were written in the first 50 seconds, and about 1.1e8 in the first 2.5 minutes. The thread stayed in this state after that.

Application thread, RUNNABLE, about one core. Top of the dump, 50 seconds into the loop:

com.azure.core.util.logging.LoggingEventBuilder.log(LoggingEventBuilder.java:208)
com.azure.core.amqp.implementation.handler.Handler.lambda$close$0(Handler.java:157)
reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:32)
com.azure.core.amqp.implementation.handler.Handler.close(Handler.java:154)
com.azure.core.amqp.implementation.ReactorConnection.closeConnectionWork(ReactorConnection.java:627)
com.azure.core.amqp.implementation.ReactorConnection.lambda$closeAsync$2(ReactorConnection.java:595)
...
com.azure.core.amqp.implementation.ReactorConnection$ReactorExceptionHandler.onConnectionError(ReactorConnection.java:760)
com.azure.core.amqp.implementation.ReactorConnection.lambda$getOrCreateConnection$2(ReactorConnection.java:716)
...
com.azure.core.amqp.implementation.ReactorDispatcher.signalWorkQueue(ReactorDispatcher.java:160)
com.azure.core.amqp.implementation.ReactorDispatcher.invoke(ReactorDispatcher.java:111)
com.azure.core.amqp.implementation.ProtonSession.open(ProtonSession.java:177)

reactor-executor-3, BLOCKED entering the same monitor, for the whole interval. CPU time flat. Lock owner is the application thread above.

com.azure.core.amqp.implementation.ReactorConnection.getOrCreateConnection(ReactorConnection.java:688)
...
reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27)
com.azure.core.amqp.implementation.handler.Handler.onNext(Handler.java:103)
com.azure.core.amqp.implementation.handler.ConnectionHandler.onConnectionRemoteOpen(ConnectionHandler.java:335)
org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:291)
com.azure.core.amqp.implementation.ReactorExecutor.run(ReactorExecutor.java:91)

Line numbers are azure-core-amqp 2.11.3. The same structure is still on main: close() returns true only for FAIL_NON_SERIALIZED, and both closeConnectionWork() and getOrCreateConnection() are private synchronized.

To Reproduce

Intermittent. Seen more than once on a ServiceBusProcessorClient workload that settles on an application thread pool by calling ServiceBusReceivedMessageContext.abandon() / complete() (both block()). Not reproduced on demand.

Code Snippet

Application side, one call per message, off the reactor thread:

messageContext.abandon(); // ServiceBusReceivedMessageContext.abandon -> receiverClient.abandon(...).block()

What the SDK then does, from 2.11.3:

// Handler.close — only FAIL_NON_SERIALIZED retries, and it retries without a bound
endpointStates.emitNext(EndpointState.CLOSED, (signalType, emitResult) -> {
    if (emitResult == Sinks.EmitResult.FAIL_NON_SERIALIZED) {
        logger.atInfo()...log("Could not emit closed endpoint state. Non-serial access. Retrying.");
        return true;
    }
    return false;
});

// Reactor's InternalManySink.emitNext
for (;;) {
    Sinks.EmitResult emitResult = tryEmitNext(value);
    if (emitResult.isSuccess()) {
        return;
    }
    if (failureHandler.onEmitFailure(SignalType.ON_NEXT, emitResult)) {
        continue;
    }
    // ...
}

// ReactorConnection.closeAsync — caller runs closeConnectionWork when the dispatcher rejects the schedule
if (dispatcher != null) {
    try {
        dispatcher.invoke(() -> closeConnectionWork());
    } catch (IOException | RejectedExecutionException e) {
        closeConnectionWork(); // synchronized, calls handler.close()
    }
}

ReactorDispatcher.signalWorkQueue is what raises the error that enters onConnectionError. A ClosedChannelException on the proton IO pipe, before the reactor is marked closed, becomes:

connectionId[...] IO Sink was interrupted before reactor closed.

Timeline from client logs. Host name omitted. t0 is the first failed abandon().

reactor-executor-3 is still inside the onConnectionRemoteOpen emit when close() emits CLOSED. tryEmitNext returns FAIL_NON_SERIALIZED. The handler returns true, so emitNext spins, still inside synchronized closeConnectionWork(). The reactor thread cannot leave its own emit, because a downstream subscriber is blocked on that same monitor in getOrCreateConnection().

No log in the window matched amqp:connection:forced, a link detach error, a timeout, or a transport error condition.

Expected behavior

Handler.close() should stop retrying. A failed CLOSED emit should not hold the ReactorConnection monitor or pin a reactor thread. Connection recovery should be able to finish, and abandon() / complete() should complete or fail instead of livelocking.

Setup (please complete the following information):

• OS: Linux
• Library/Libraries: com.azure:azure-messaging-servicebus:7.17.17, which pulls com.azure:azure-core-amqp:2.11.3. reactor-core 3.7.x (InternalManySink.emitNext still busy-loops when the handler returns true)
• Java version: HotSpot, JDK 15+ lambda frame form ($$Lambda/0x...)
• Frameworks: none. ServiceBusProcessorClient, blocking abandon() / complete() from an application executor
• Also checked: azure-messaging-servicebus 7.18.1 / azure-core-amqp 2.12.2, and current main. Handler.close, ReactorConnection.closeConnectionWork, and getOrCreateConnection are unchanged. 7.18.1 still calls .block() from ServiceBusReceivedMessageContext.abandon() and complete()

Information Checklist

• Bug Description Added
• Repro Steps Added
• Setup information Added