What causes "Receive interrupted" in apache/pulsar?

Handle PulsarClientException in the receive loop and check Thread.currentThread().isInterrupted() to exit cleanly.

apache/pulsar · warning · PulsarClientException

Receive interrupted

Error message

Receive interrupted

What it means

V5ReceiveQueue.take() blocks on receiveAsync().get(). If the waiting thread is interrupted while blocked, the interrupt status is restored and a PulsarClientException with the fixed message "Receive interrupted" is thrown. ExecutionException causes are unwrapped and rethrown via unwrap(e) instead.

Solutions

Handle PulsarClientException in the receive loop and check Thread.currentThread().isInterrupted() to exit cleanly.
Use poll(Duration) with a timeout instead of indefinite take() so shutdown can be detected without interrupts.
Ensure a producer is actually publishing to the topic; idle topics make take() wait indefinitely until interrupted.
Prefer receiveMulti(maxMessages, timeout) for batch consumption with bounded waiting.

Example fix

// before
while (running) {
    Message<String> msg = queue.take(); // throws on interrupt
    handle(msg);
}
// after
while (running) {
    Message<String> msg = queue.poll(Duration.ofSeconds(1));
    if (msg == null) continue;
    handle(msg);
}

Defensive patterns

Strategy: try-catch

Validate before calling

if (Thread.currentThread().isInterrupted()) {
    // don't enter a blocking take with a pending interrupt
    throw new PulsarClientException("interrupted before take");
}

Try / catch

try {
    Message<T> msg = queue.take();
} catch (PulsarClientException e) {
    if (Thread.currentThread().isInterrupted()) return; // shutdown path
    throw e;
}

Prevention

Use poll(Duration) instead of take() in shutdown-sensitive loops
Stop consumer threads with a flag plus bounded waits, not interrupts
Confirm producers are active so take() doesn't wait indefinitely
Always restore-and-honor the interrupt status in receive loops

When it happens

Trigger: Calling take() and having the thread interrupted while waiting for the next message — consumer closed, executor shutdown, or another thread calling interrupt().

Common situations: Consumer loop threads interrupted during application shutdown; watchdogs interrupting threads that appear stuck waiting for messages that never arrive (no producers publishing); test timeouts cancelling receive loops.

Related errors

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

Appendix: source

Thrown at pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueue.java:272

            Message<T> m;
            while (batch.size() < max && (m = buffer.poll()) != null) {
                batch.add(m);
            }
            approxBufferSize = buffer.size();
            maybeResumeProducers();
            done.complete(null);
        });
        return done;
    }

    // --- Blocking views, for the synchronous receive() API. Block only the caller's thread. ---

    Message<T> take() throws PulsarClientException {
        try {
            return receiveAsync().get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new PulsarClientException("Receive interrupted", e);
        } catch (ExecutionException e) {
            throw unwrap(e);
        }
    }

    Message<T> poll(Duration timeout) throws PulsarClientException {
        try {
            return receiveAsync(timeout).get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new PulsarClientException("Receive interrupted", e);
        } catch (ExecutionException e) {
            throw unwrap(e);
        }
    }

    List<Message<T>> receiveMulti(int maxMessages, Duration timeout) throws PulsarClientException {
        try {

View on GitHub (pinned to 820761864e)