[Bug][client] Unpooled connections (connectionsPerBroker=0) still being established when the client closes are leaked

[lhotari]

Search before reporting

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

User environment

Pulsar Java client, current master (b27d3ad). Found by code inspection during the review of #26723.

Issue Description

When connection pooling is disabled (connectionsPerBroker(0)), a connection that is still being established when the client closes can leak.

• In ConnectionPool.getConnection(), maxConnectionsPerHosts == 0 goes straight to createConnection(...) without storing the future in pool.
• ConnectionPool.closeAllConnections() only goes through pool.values(), so it doesn't see these in-flight unpooled connections.
• When such a connection completes after its producer or consumer has been closed, the closed-state early return in ProducerImpl.connectionOpened() and ConsumerImpl.connectionOpened() doesn't release the ClientCnx. For unpooled connections, ConnectionPool.releaseConnection() is what closes the channel.

The channel stays open until its Netty event loop shuts down, because Netty closes all registered channels at that point. With a client-owned event loop this happens when the client closes. With an event loop shared through PulsarClientSharedResources, it happens only when the shared resources are closed, so the connection (and the broker side of it) can stay open for much longer.

This doesn't depend on the DNS resolver: a client that closes while channel.connect() is still in flight, after DNS resolution has finished, ends up the same way. See the review discussion on #26723.

Error messages

None; the leaked connection stays open without any error.

Reproducing the issue

Expected sequence based on code inspection; not yet reproduced with a test:

1. Create PulsarClientSharedResources with a shared event loop, and create a client that uses it with connectionsPerBroker(0).
2. Start creating a producer or consumer so that a new connection is being established (DNS resolution or TCP connect in flight).
3. Close the client before the connection completes.
4. The connection completes on the shared event loop and stays open until the shared resources are closed.

Additional information

Possible fixes:

• Track pending and active unpooled connection futures in ConnectionPool so that closeAllConnections() closes them too.
• And/or release (close) the ClientCnx in the closed-state early-return path of connectionOpened() when pooling is disabled.

Are you willing to submit a PR?

• I'm willing to submit a PR!