apache/pulsar · warning · PulsarClientException

Close interrupted

Error message

Close interrupted

What it means

ScalableStreamConsumer.close() blocks on closeAsync().get(); if the thread is interrupted while waiting, the interrupt flag is restored and PulsarClientException("Close interrupted") is thrown. The underlying close of all stream segments may still continue in the background.

Source

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

        if (!drainingConsumers.containsKey(segmentId)) {
            return;
        }
        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);
    }

View on GitHub (pinned to 820761864e)

Solutions

1. Close consumers before shutting down executors / interrupting threads
2. Use closeAsync().get(timeout, TimeUnit) instead of interruption for deadlines
3. Catch the exception and verify Thread.interrupted() to confirm this scenario
4. If interrupted, call closeAsync() again from a non-interrupted thread to confirm completion

Example fix

// before
// in shutdown hook, with pool already shutdownNow()
streamConsumer.close();
// after
streamConsumer.closeAsync().get(30, TimeUnit.SECONDS);
executor.shutdownNow(); // interrupt only after close finished

try {
    streamConsumer.close();
} catch (PulsarClientException e) {
    if ("Close interrupted".equals(e.getMessage())) {
        log.warn("stream close interrupted; segments may still be closing");
        streamConsumer.closeAsync(); // finish in background
    }
}

When it happens

Trigger: Interruption of the thread blocked in closeAsync().get(): ExecutorService.shutdownNow() during app shutdown, future.cancel(true), or timeout frameworks that interrupt threads.

Common situations: SIGTERM handlers interrupting shutdown threads before the multi-segment close finishes; shutdown timeouts that call shutdownNow(); interrupt-based cancellation in worker loops.

Related errors

• Close interrupted
• Close interrupted
• Close interrupted
• ${cause}
• ${cause}

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