[BUG] Possible memory leak azure-sdk-bom 1.3.6 with netty 4.1.134-FINAL

[riconstantin]

Describe the bug
We have discovered a possible memory leak with a very specific use-case: we use azure communication only when an application is deployed and never after. Because of this, we run into Azure Service Bus AMQP idle-timeout reconnects. From our analysis on some memory dumps from before the app gets OOM killed, we noticed that every reconnect allocated ~67–134 MB of native memory that the allocator never returned to the OS. Over 6 days and dozens of reconnect cycles, this accumulated to ~1.8–2.0 GB of invisible off-heap memory — exhausting the 6 GiB container limit.

Exception or Stack Trace

The specific chain:

Azure Service Bus broker: 300-second AMQP idle timeout
↓
broker sends amqp:connection:forced
↓
Azure SDK (Reactor/Netty): ReactorSession + RequestResponseChannel errors
↓
Netty PooledByteBufAllocator allocates new native Chunk(s) (~67–134 MB)
via sun.misc.Unsafe.allocateMemory() — bypasses ALL JVM memory metrics
↓
Old chunks returned to pool arena, but native OS pages NEVER freed
↓
RSS grows by ~67–134 MB per reconnect, permanently
↓
After ~11 reconnects: RSS at 97% of 6 GiB limit
↓
13:16:46Z: minor GC burst + CPU spike on both pods
↓
RSS crosses 6 GiB → kernel OOM killer → SIGKILL on both replicas

Setup (please complete the following information):

• Library/Libraries: azure-sdk-bom 1.3.6, netty 4.1.134-FINAL
• Java version: 21
• App Server/Environment: Openshift
• Frameworks: SpringBoot 3.5.8

Additional context
I can provide (if needed) memory dump + the analysis.

Information Checklist
Kindly make sure that you have added all the following information above and checkoff the required fields otherwise we will treat the issuer as an incomplete report

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

[schiemon]

This looks like a bug in ConnectionHandler. On a service-initiated close that carries an error condition (e.g. the amqp:connection:forced frame the broker sends on the 300s idle timeout), the closure is never signaled to subscribers, so the old connection and its transport are never disposed and leak on every reconnect.

Root cause

ConnectionHandler.onConnectionRemoteClose emits the CLOSED endpoint state only on the graceful path; on the error path it calls notifyErrorContext instead:

azure-sdk-for-java/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java

Lines 355 to 365 in 8f266e9

	public void onConnectionRemoteClose(Event event) {
	final Connection connection = event.getConnection();
	final ErrorCondition error = connection.getRemoteCondition();
	logErrorCondition("onConnectionRemoteClose", connection, error);
	if (error == null) {
	onNext(connection.getRemoteState());
	} else {
	notifyErrorContext(connection, error);
	}
	}

But notifyErrorContext early-returns when remoteState == CLOSED, which is always true once a remote close frame has been received. So on amqp:connection:forced neither a state nor an error is emitted:

azure-sdk-for-java/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java

Lines 389 to 404 in 8f266e9

	private void notifyErrorContext(Connection connection, ErrorCondition condition) {
	if (connection == null || connection.getRemoteState() == EndpointState.CLOSED) {
	return;
	}
	if (condition == null) {
	throw logger
	.logExceptionAsError(new IllegalStateException("notifyErrorContext does not have an ErrorCondition."));
	}
	// if the remote-peer abruptly closes the connection without issuing close frame issue one
	final Throwable exception = ExceptionUtil.toException(condition.getCondition().toString(),
	condition.getDescription(), getErrorContext());
	onError(exception);
	}

That endpoint-state flux in ReactorConnection is what drives connection teardown. ReactorConnection only calls closeAsync() from onErrorResume / doOnComplete:

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

Lines 176 to 195 in 8f266e9

	this.endpointStates
	= this.handler.getEndpointStates().takeUntilOther(shutdownSignalSink.asMono()).map(state -> {
	logger.atVerbose().addKeyValue("state", state).log("getConnectionState");
	return AmqpEndpointStateUtil.getConnectionState(state);
	}).onErrorResume(error -> {
	if (setDisposed()) {
	logger.verbose("Disposing of active sessions due to error.");
	return closeAsync(new AmqpShutdownSignal(false, false, error.getMessage())).then(Mono.error(error));
	} else {
	return Mono.error(error);
	}
	}).doOnComplete(() -> {
	if (setDisposed()) {
	logger.verbose("Disposing of active sessions due to connection close.");
	closeAsync(new AmqpShutdownSignal(false, false, "Connection handler closed.")).subscribe();
	}
	}).cache(1);
	this.subscriptions = Disposables.composite(
	this.endpointStates.subscribe(null, e -> logger.warning("Connection endpoint state signaled error .", e)));

With no signal, closeAsync() is not called and the transport is never unbound/freed.

This is a regression from f2eb437; previously onNext(connection.getRemoteState()) was called unconditionally.

Here is a reproducer (a test in ConnectionHandlerTest):

    @Test
    void onConnectionRemoteCloseWithErrorSignalsClosure() {
        // Arrange: the broker forces the connection closed with an 'amqp:connection:forced' error condition.
        final ErrorCondition errorCondition
            = new ErrorCondition(AmqpErrorCode.CONNECTION_FORCED, "The connection was closed due to idle timeout.");

        final Connection connection = mock(Connection.class);
        // A remote close frame was received, so the remote endpoint state is CLOSED and carries the error condition.
        when(connection.getRemoteState()).thenReturn(EndpointState.CLOSED);
        when(connection.getRemoteCondition()).thenReturn(errorCondition);
        when(connection.getHostname()).thenReturn(HOSTNAME);

        final Event closeEvent = mock(Event.class);
        when(closeEvent.getConnection()).thenReturn(connection);

        final List<EndpointState> states = new ArrayList<>();
        final AtomicReference<Throwable> error = new AtomicReference<>();
        final AtomicBoolean completed = new AtomicBoolean();
        final Disposable subscription
            = handler.getEndpointStates().subscribe(states::add, error::set, () -> completed.set(true));

        // Act
        handler.onConnectionRemoteClose(closeEvent);
        subscription.dispose();

        // Assert: subscribers must be told the connection is closed
        final boolean signaledClosure = states.contains(EndpointState.CLOSED) || completed.get() || error.get() != null;
        
        assertTrue(signaledClosure,
            "onConnectionRemoteClose with an error condition must signal connection closure to subscribers; "
                + "otherwise the connection/transport is never disposed and leaks on every reconnect. "
                + "Observed states=" + states + ", completed=" + completed.get() + ", error=" + error.get());
    }

On current main (8f266e9) this fails with

[ERROR] Failures: 
[ERROR]   ConnectionHandlerTest.onConnectionRemoteCloseWithErrorSignalsClosure:335 states=[UNINITIALIZED], completed=false, error=null ==> expected: <true> but was: <false>

[alzimmermsft]

Hi @riconstantin, @schiemon, you mentioned Netty 4.1.134.Final is being used, the resolved version in azure-sdk-bom 1.3.6 should be 4.1.132.Final, is the Netty version being overridden manually (it's fine if it is, just looking for more information).

Netty released 4.1.135.Final, which we just begun releasing new versions of Azure SDKs using it, which has a few CVE fixes for memory leaks on their side. Would it be possible to test out using that version of Netty to see if it resolves this issue?

[riconstantin]

Hi, and thanks for looking into this issue. It seems I got confused by what mvnrepository was saying and didn't check thoroughly what is actually used. I have checked again, and the correct versions used are: azure-core-http-netty 1.16.3, netty-* 4.1.128.FINAL, azure-core-ampq 2.11.3 (mentioned by @schiemon). I'll try using netty 4.1.135.Final, in the meantime, I implemented sending/receiving a Noop message in order to keep the connection alive and I'm monitoring results.