{"record":{"id":"a13435b43d9e506d","repo":"apache/pulsar","slug":"receive-interrupted","errorCode":null,"errorMessage":"Receive interrupted","messagePattern":"Receive interrupted","errorType":"exception","errorClass":"PulsarClientException","httpStatus":null,"severity":"warning","filePath":"pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueue.java","lineNumber":272,"sourceCode":"            Message<T> m;\n            while (batch.size() < max && (m = buffer.poll()) != null) {\n                batch.add(m);\n            }\n            approxBufferSize = buffer.size();\n            maybeResumeProducers();\n            done.complete(null);\n        });\n        return done;\n    }\n\n    // --- Blocking views, for the synchronous receive() API. Block only the caller's thread. ---\n\n    Message<T> take() throws PulsarClientException {\n        try {\n            return receiveAsync().get();\n        } catch (InterruptedException e) {\n            Thread.currentThread().interrupt();\n            throw new PulsarClientException(\"Receive interrupted\", e);\n        } catch (ExecutionException e) {\n            throw unwrap(e);\n        }\n    }\n\n    Message<T> poll(Duration timeout) throws PulsarClientException {\n        try {\n            return receiveAsync(timeout).get();\n        } catch (InterruptedException e) {\n            Thread.currentThread().interrupt();\n            throw new PulsarClientException(\"Receive interrupted\", e);\n        } catch (ExecutionException e) {\n            throw unwrap(e);\n        }\n    }\n\n    List<Message<T>> receiveMulti(int maxMessages, Duration timeout) throws PulsarClientException {\n        try {","sourceCodeStart":254,"sourceCodeEnd":290,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueue.java#L254-L290","documentation":"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.","triggerScenarios":"Calling take() and having the thread interrupted while waiting for the next message — consumer closed, executor shutdown, or another thread calling interrupt().","commonSituations":"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.","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."],"exampleFix":"// before\nwhile (running) {\n    Message<String> msg = queue.take(); // throws on interrupt\n    handle(msg);\n}\n// after\nwhile (running) {\n    Message<String> msg = queue.poll(Duration.ofSeconds(1));\n    if (msg == null) continue;\n    handle(msg);\n}","handlingStrategy":"try-catch","validationCode":"if (Thread.currentThread().isInterrupted()) {\n    // don't enter a blocking take with a pending interrupt\n    throw new PulsarClientException(\"interrupted before take\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    Message<T> msg = queue.take();\n} catch (PulsarClientException e) {\n    if (Thread.currentThread().isInterrupted()) return; // shutdown path\n    throw e;\n}","preventionTips":["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"],"tags":["receive","interrupt","blocking","consumer"],"backgroundTag":"thread-interrupted","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}