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

Close interrupted

Error message

Close interrupted

What it means

ScalableCheckpointConsumer.close() blocks on closeAsync().get(); if the waiting thread is interrupted, it restores the interrupt flag and throws a PulsarClientException with the fixed message 'Close interrupted'. This means the caller was interrupted while waiting for the consumer's async close to finish — the close itself may still be proceeding in the background.

Source

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

    @Override
    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);

View on GitHub (pinned to 820761864e)

Solutions

1. Determine what interrupted the thread; the library already re-sets the interrupt flag, so honor it in your cleanup path
2. Avoid interrupting threads that own consumer close; prefer letting closeAsync() complete and attaching callbacks
3. If you must bound close time, use closeAsync() with orTimeout instead of interrupting the blocking get()

Example fix

// before
consumer.close(); // blocking, interrupt-sensitive
// after
consumer.closeAsync().orTimeout(30, TimeUnit.SECONDS).join();

try {
    consumer.close();
} catch (PulsarClientException e) {
    if (Thread.currentThread().isInterrupted()) {
        log.warn("Consumer close interrupted");
    } else {
        throw e;
    }
}

When it happens

Trigger: Calling ScalableCheckpointConsumer.close() from a thread that gets interrupted while blocked in closeAsync().get() — executor shutdownNow(), request-timeout interrupts, or application shutdown hooks.

Common situations: Checkpoint/recovery frameworks interrupting worker threads; container shutdown during a long-running close; cancellation frameworks interrupting tasks that own the consumer.

Related errors

• Consumer was not connected
• Cannot use receive() when a listener has been set
• Can't use receive with timeout, if the queue size is 0
• Consumer already closed
• Topic was terminated

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