apache/pulsar · error · org.apache.pulsar.client.impl.v5.PulsarClientException

${cause}

Error message

${cause}

What it means

ScalableCheckpointConsumer.close() blocks on closeAsync().get() and wraps any failure of the asynchronous close in a PulsarClientException. When the future completes exceptionally, ExecutionException is caught and the underlying cause is rethrown as `new PulsarClientException(e.getCause())`, so `${cause}` is the message of the real underlying failure (e.g. broker disconnect, already-closed consumer).

Source

Thrown at pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java:225

    public Checkpoint checkpoint() {
        Map<Long, org.apache.pulsar.client.api.MessageId> positions = new HashMap<>(lastReceivedPositions);
        return new CheckpointV5(positions);
    }

    @Override
    public AsyncCheckpointConsumer<T> async() {
        return asyncView;
    }

    @Override
    public void close() throws PulsarClientException {
        try {
            closeAsync().get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new PulsarClientException("Close interrupted", e);
        } catch (ExecutionException e) {
            throw new PulsarClientException(e.getCause());
        }
    }

    // --- Async internals ---

    CompletableFuture<Message<T>> receiveAsync() {
        return receiveQueue.receiveAsync().thenApply(this::advanceCheckpoint);
    }

    CompletableFuture<Message<T>> receiveAsync(Duration timeout) {
        return receiveQueue.receiveAsync(timeout).thenApply(this::advanceCheckpoint);
    }

    CompletableFuture<List<Message<T>>> receiveMultiAsync(int maxMessages, Duration timeout) {
        return receiveQueue.receiveMultiAsync(maxMessages, timeout).thenApply(batch -> {
            batch.forEach(this::advanceCheckpoint);
            return batch;
        });

View on GitHub (pinned to 820761864e)

Solutions

1. Inspect the cause chain of the thrown PulsarClientException (getCause()) to find the real close failure
2. Verify broker connectivity and that the broker is up before/while closing
3. Avoid closing the same consumer twice; guard with an isClosed flag or try-finally
4. Retry close() after a transient network failure; consider closeAsync() for finer control

Example fix

// before
consumer.close();
// after
try {
    consumer.close();
} catch (PulsarClientException e) {
    log.warn("close failed: {}", e.getCause(), e);
    // inspect e.getCause() for the underlying failure and retry if transient
}

// no pre-check API; ensure client is connected and consumer not already closed
if (consumer == null || alreadyClosed) { skipClose(); }

try {
    consumer.close();
} catch (PulsarClientException e) {
    Throwable cause = e.getCause();
    log.warn("close failed: {}", cause, cause);
    if (isTransient(cause)) retryClose(consumer);
}

When it happens

Trigger: Calling close() when the underlying async close future completes exceptionally — e.g. a segment consumer failed to close, the broker rejected the close (ConsumerBusy/AlreadyClosed), or a network error occurred while unsubscribing/checkpoint state was being cleaned up.

Common situations: Shutting down an application while the broker is unreachable; double-closing a consumer; closing during a broker restart; timeouts while closing many segments of a scalable checkpoint consumer.

Related errors

• ${cause}
• (wraps underlying failure cause)
• Consumer already closed
• Topic was terminated
• ServiceUrlProvider has already been initialized

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/77e1e1a5906e6316. Report an issue: GitHub.