[fix][client] Fail getPartitionsForTopic on a closed client instead of never completing - #27491

Motivation

Problem. Once a PulsarClient is closed, getPartitionsForTopic() returns a future that never completes: a
caller blocked on get() without a timeout hangs, and an asynchronous caller is never called back. Producer,
consumer and reader creation fail at once with AlreadyClosedException after close(), because
createProducerAsync(), subscribeAsync() and createReaderAsync() check the client state.
getPartitionedTopicMetadata(String, boolean, boolean), which getPartitionsForTopic() calls, doesn't. It is also
called by ConsumerBuilderImpl.subscribeAsync() for the retry and dead-letter topics of a consumer with
enableRetry(true), before the state check of the client's subscribeAsync() (so that subscribe() hangs on a
closed client too), by the partition auto-update of MultiTopicsConsumerImpl and PartitionedProducerImpl, and by
the broker when it checks the remote clusters before auto-creating a topic, through its replication clients.

On a closed client, the lookup itself never completes. The connection pool resolves the broker address with
Netty's InetSocketAddressResolver, which adds a listener to a promise of the client's event loop. That loop has
terminated, so Netty can't submit the notification and drops it:

ERROR i.n.u.c.D.rejectedExecution - Failed to submit a listener notification task. Event loop shut down?
java.util.concurrent.RejectedExecutionException: event executor terminated
	...
	at io.netty.util.concurrent.DefaultPromise.addListener(DefaultPromise.java:199)
	at io.netty.resolver.InetSocketAddressResolver.doResolveAll(InetSocketAddressResolver.java:71)
	at io.netty.resolver.AbstractAddressResolver.resolveAll(AbstractAddressResolver.java:158)
	at org.apache.pulsar.client.impl.ConnectionPool.resolveName(ConnectionPool.java:457)
	...
	at org.apache.pulsar.client.impl.BinaryProtoLookupService.getPartitionedTopicMetadataAsync(BinaryProtoLookupService.java:267)
	...
	at org.apache.pulsar.client.impl.PulsarClientImpl.getPartitionsForTopic(PulsarClientImpl.java:1994)

A lookup in flight when the client closes is left pending too, by the retry loop. Closing the connection fails the
request with ConnectException ("Disconnected from server at ..."), which is retriable, so the lookup is retried
after a backoff on the client's scheduled executor. That happens while the client is closing, before it stops its
executors; it then stops them with shutdownNow(), which discards the retry.

Example.

PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
client.close();
// Never completes: TimeoutException
client.getPartitionsForTopic("persistent://public/default/t", true).get(5, TimeUnit.SECONDS);

The code path:

public CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(
        String topic, boolean metadataAutoCreationEnabled, boolean useFallbackForNonPIP344Brokers) {
    // No state check
    CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
    ...
    getPartitionedTopicMetadata(topicName, backoff, opTimeoutMs, metadataFuture, ...);
    return metadataFuture;
}

private void getPartitionedTopicMetadata(TopicName topicName, Backoff backoff, ...) {
    // lookup -> ConnectionPool.getConnection() -> resolveName(): never completes on a closed client
    CompletableFuture<PartitionedTopicMetadata> queryFuture = lookup.getPartitionedTopicMetadata(topicName, ...);
    queryFuture.thenAccept(future::complete).exceptionally(e -> {
        ...
        if (nextDelay <= 0 || isLookupThrottling) {
            ...
            future.completeExceptionally(e);
            return null;
        }
        previousExceptionCount.getAndIncrement();
        // While the client is closing, its shutdownNow() discards this retry
        ((ScheduledExecutorService) scheduledExecutorProvider.getExecutor()).schedule(() -> {
            ...
        }, nextDelay, TimeUnit.MILLISECONDS);
        return null;
    });
}

Modifications

• getPartitionedTopicMetadata(String, boolean, boolean) fails with AlreadyClosedException once the client is
  closing or closed. It checks isClosed() rather than state != Open as the producer and consumer creation do:
  the transaction coordinator client looks up its assign topic from the client's constructor, before the client is
  Open, so with state != Open every client built with enableTransaction(true) would fail to start.
• A lookup that fails with a retriable error while the client is closing fails with AlreadyClosedException
  instead of being retried, since the close would discard the retry.

// Not state != Open: the transaction coordinator client calls this from the constructor, before the client is
// Open
if (isClosed()) {
    return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed"));
}
...
// A closing client would discard the retry when it shuts down its executor, leaving the future pending
if (isClosed()) {
    future.completeExceptionally(new PulsarClientException.AlreadyClosedException("Client already closed"));
    return null;
}

A non-retriable error, or one that comes when the lookup has no time left, still fails the lookup with that error,
as before.

Since #27473, shutdown() on its own marks the client closed as well, so getPartitionsForTopic() after
shutdown() fails with AlreadyClosedException too (checked locally with this change on top of #27473's).

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• PulsarClientImplTest#testGetPartitionsForTopicFailsOnClosedClient: getPartitionsForTopic() on a closed client
  must fail with AlreadyClosedException within 10 seconds. On master it times out.
• PulsarClientImplTest#testGetPartitionsForTopicFailsWhenClientClosesDuringLookup: the lookup request fails with a
  retriable ConnectException while the client is closing, as when the client closes the connection it was sent
  on. getPartitionsForTopic() must fail with AlreadyClosedException. On master it times out: the retry is
  scheduled, then discarded when the close stops the executor.
• PulsarClientImplTest (23) and TransactionClientCloseTest (7), whose clients have transactions enabled and
  so look up the coordinator's assign topic from the client's constructor, pass.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment