Visitar URL original
[fix][client] Mark PulsarClient closed on shutdown() so new producers and consumers fail fast by merlimat · Pull Request #27473 · apache/pulsar · GitHub
Skip to content

[fix][client] Mark PulsarClient closed on shutdown() so new producers and consumers fail fast - #27473

Merged
merlimat merged 1 commit into
apache:masterfrom
merlimat:mmerli/client-shutdown-closed-state
Nov 9, 2026
Merged

merlimat merged 1 commit into
apache:masterfrom
merlimat:mmerli/client-shutdown-closed-state

Conversation

@merlimat

@merlimat merlimat commented Nov 9, 2026

Copy link
Copy Markdown
Contributor

Motivation

PulsarClient.shutdown() is documented as an immediate shutdown: it releases all the resources without waiting for
ongoing operations. PulsarClientImpl.shutdown() releases them (lookup service, transaction coordinator client, DNS
resolver, connection pool, event loop, timer, executors, authentication), but never changes the client's state. Only
closeAsync() does: it moves the state from Open to Closing, closes the producers and consumers, calls
shutdown(), and then sets Closed.

So after a direct shutdown() the client still says it is open, and the state checks let new work through to the
released resources:

PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
client.shutdown();

client.isClosed();                                                     // false
client.newProducer().topic("persistent://public/default/t").create();  // never returns
// PulsarClientImpl.createProducerAsync(); subscribeAsync() and createReaderAsync() have the same check
if (state.get() != State.Open) {
    return FutureUtil.failedFuture(
            new PulsarClientException.AlreadyClosedException("Client already closed : state = " + state.get()));
}

The creation goes on to the partitioned-metadata lookup on the closed lookup service and never completes. Producers,
consumers, readers and table views all hang this way, where after close() they fail with AlreadyClosedException
right away. In AdminApiTest, the test methods that used the shared client after testNamespaceUnloadBundle had
called pulsarClient.shutdown() hung until their 300 s TestNG timeouts instead of failing at once.

Modifications

shutdown() now sets the state itself, the way closeAsync() does around it:

@Override
public void shutdown() throws PulsarClientException {
    // Fail the producers, consumers and readers created from now on, as closeAsync() does before it gets here:
    // they would wait for ever on the resources released below.
    state.compareAndSet(State.Open, State.Closing);
    try {
        ...
    } finally {
        state.set(State.Closed);
    }
}
  • An open client is Closing while shutdown() releases its resources, and Closed afterwards, also when a release
    step fails (as closeAsync() sets Closed when shutdown() throws).
  • closeAsync() keeps its Open → Closing → shutdown() → Closed sequence: shutdown() leaves its Closing
    as is, and the Closed it sets afterwards is already there.
  • isClosed() is true after shutdown(), as its javadoc (@see #shutdown()) implies. The code that checks
    isClosed() to stop retrying, such as the v5 DAG watch reconnects and the table view's tail reads, now stops after
    shutdown() as it does after close().

Compatibility:

  • Creating a producer, consumer, reader or table view on a shut-down client fails with AlreadyClosedException at
    once, as after close(), instead of never completing.
  • closeAsync() after shutdown() fails with AlreadyClosedException, like a second closeAsync(), and close()
    returns without doing anything, as it ignores that exception. Before, they ran the close sequence on the shut-down
    client: closing the producers and consumers created before the shutdown, then running shutdown() a second time.
    Every caller of shutdown() in this repository drops the client afterwards (the broker's replication and namespace
    clients, PulsarClientV5.shutdown(), the test base classes), and none of them closes it.
  • Unchanged: shutdown() still doesn't close the producers and consumers created before it, and a second shutdown()
    still runs the release steps again. Transactions already failed fast after shutdown(), which closes the transaction
    coordinator client (CoordinatorClientStateException).

V5ClientLifecycleTest.testShutdownDelegatesToV4 described v4's shutdown() as deliberately not setting the client's
state to Closed; that sentence is removed. The behavior dates back to the first version of the client.

Verifying this change

This change added tests and can be verified as follows:

  • PulsarClientImplTest.testShutdownClosesClient shuts a client down, then creates a producer, a consumer, a reader
    and a table view, each of which must fail with AlreadyClosedException within 10 s, and checks that isClosed() is
    true and that close() doesn't throw. Without the change, the producer creation doesn't complete within the 10 s
    (TimeoutException).

Also passing locally: PulsarClientImplTest (22 tests), the other pulsar-client unit tests (except
OAuth2IdpTlsFrameworkClientTest, whose WireMock server couldn't bind port 8080, taken by another process on the
machine), the pulsar-client-v5 unit tests, V5ClientLifecycleTest, and ./gradlew quickCheck.

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

… and consumers fail fast

PulsarClientImpl.shutdown() released the client's resources but left its
state Open: only closeAsync() set Closed, after calling shutdown(). So
isClosed() stayed false, and the producers, consumers, readers and table
views created afterwards passed the state checks and never completed,
instead of failing with AlreadyClosedException as they do after close().

shutdown() now moves an open client to Closing before it releases anything,
as closeAsync() does, and leaves it Closed. closeAsync() keeps its
Open -> Closing -> shutdown() -> Closed sequence.
@merlimat
merlimat merged commit 31f9dbf into apache:master Nov 9, 2026
44 checks passed
lhotari pushed a commit that referenced this pull request Nov 9, 2026
… and consumers fail fast (#27473)

(cherry picked from commit 31f9dbf)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants