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.

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)

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.