apache/pulsar · error · PulsarClientException

(wraps underlying failure cause)

Error message

(wraps underlying failure cause)

What it means

close() converts a failed closeAsync() future into `new PulsarClientException(e.getCause())`; the thrown message is the underlying cause's message. For a stream consumer this typically means one of the segment consumers failed to close (broker error, connection loss, AlreadyClosed).

Source

Thrown at pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java:388

        }
        pendingDrainAcks.computeIfAbsent(segmentId, __ -> new ConcurrentLinkedQueue<>())
                .add(ackFuture.exceptionally(ex -> null));
    }

    @Override
    public AsyncStreamConsumer<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();
    }

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

    CompletableFuture<List<Message<T>>> receiveMultiAsync(int maxNumMessages, Duration timeout) {
        return receiveQueue.receiveMultiAsync(maxNumMessages, timeout);
    }

    CompletableFuture<Void> closeAsync() {

View on GitHub (pinned to 820761864e)

Solutions

1. Inspect e.getCause() for the concrete broker/client error
2. Retry close after transient connectivity issues; the close is idempotent for already-closed segments
3. Check broker availability before shutdown; sequence shutdown: consumers → producers → client
4. Prefer closeAsync() and log each segment's failure with handle((c, ex) -> ...)

Example fix

// before
streamConsumer.close();
// after
try {
    streamConsumer.close();
} catch (PulsarClientException e) {
    log.error("stream close failed: {}", e.getCause(), e);
    // retry or continue shutdown depending on cause
}

// verify broker connectivity before shutdown
// client.getPartitionsForTopic(...) or a lightweight lookup as health check

try {
    streamConsumer.close();
} catch (PulsarClientException e) {
    Throwable cause = e.getCause();
    log.error("stream close failed: {}", cause, cause);
    if (isTransient(cause)) streamConsumer.closeAsync();
}

When it happens

Trigger: closeAsync() completing exceptionally while closing the segment DAG: broker rejects close, network disconnects mid-close, or a segment consumer was already closed/failed.

Common situations: Application shutdown during broker failover; closing a stream consumer whose subscription was concurrently deleted; repeated close calls on a partially closed consumer.

Related errors

• ${cause}
• ${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/14b4ae4d893a0c10. Report an issue: GitHub.