apache/pulsar · error · PulsarClientException

Close interrupted

Error message

Close interrupted

What it means

ScalableTopicProducer.close() blocks on closeAsync().get(); if the waiting thread is interrupted, it restores the interrupt flag and throws a generic PulsarClientException with this message. It signals that producer close did not complete because the caller's thread was interrupted, not because of a broker problem.

Source

Thrown at pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java:179

            if (future.isDone() && !future.isCompletedExceptionally()) {
                max = Math.max(max, future.join().getLastSequenceId());
            }
        }
        return max;
    }

    @Override
    public AsyncProducer<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) {
            Throwable cause = e.getCause();
            if (cause instanceof PulsarClientException pce) {
                throw pce;
            }
            throw new PulsarClientException(cause);
        }
    }

    /**
     * Send a message synchronously with routing. Called by MessageBuilderV5.
     * Returns a MessageIdV5 that includes the segment ID for ack routing.
     */
    MessageIdV5 sendInternal(
            String key, T value, java.util.Map<String, String> properties,
            java.time.Instant eventTime, Long sequenceId,
            java.time.Duration deliverAfter, java.time.Instant deliverAt,
            java.util.List<String> replicationClusters,

View on GitHub (pinned to 820761864e)

Solutions

1. Avoid interrupting threads that are closing producers; use graceful shutdown (shutdown() + awaitTermination) before interrupting.
2. Check the interrupt source: the interrupt flag is re-set on the thread, so inspect Thread.interrupted() handling in the surrounding code.
3. If interruption is expected (e.g. timeouts), catch PulsarClientException and treat close as best-effort, or prefer closeAsync() with orTimeout for time-bounded shutdown.
4. Ensure close() is only called once per producer from a dedicated lifecycle thread.

Example fix

// before
producer.close(); // interrupted during shutdown
// after
CompletableFuture<Void> f = producer.closeAsync();
try {
    f.get(30, TimeUnit.SECONDS);
} catch (TimeoutException te) {
    f.cancel(true);
}

try {
    producer.close();
} catch (PulsarClientException e) {
    if (e.getCause() instanceof InterruptedException) {
        Thread.currentThread().interrupt(); // already re-set, proceed with shutdown
    } else { throw e; }
}

When it happens

Trigger: Calling close() (directly or via try-with-resources) on a ScalableTopicProducer while another thread interrupts the calling thread, e.g. during shutdown, executor termination, or task timeout.

Common situations: Application shutdown interrupting worker threads mid-close; Future/timeout cancellation; a thread-pool executor calling close() and being shut down with shutdownNow(); publisher loops cancelled in tests.

Related errors

• Interrupted while creating segment producer
• ServiceUrlProvider has already been initialized
• Authentication already closed.
• Interrupted initializing OAuth2 IdP TLS factory
• (wraps underlying failure cause)

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