{"record":{"id":"7aa30d8646b6b509","repo":"apache/pulsar","slug":"close-interrupted-7aa30d","errorCode":null,"errorMessage":"Close interrupted","messagePattern":"Close interrupted","errorType":"exception","errorClass":"org.apache.pulsar.client.impl.v5.PulsarClientException","httpStatus":null,"severity":"warning","filePath":"pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java","lineNumber":223,"sourceCode":"\n    @Override\n    public Checkpoint checkpoint() {\n        Map<Long, org.apache.pulsar.client.api.MessageId> positions = new HashMap<>(lastReceivedPositions);\n        return new CheckpointV5(positions);\n    }\n\n    @Override\n    public AsyncCheckpointConsumer<T> async() {\n        return asyncView;\n    }\n\n    @Override\n    public void close() throws PulsarClientException {\n        try {\n            closeAsync().get();\n        } catch (InterruptedException e) {\n            Thread.currentThread().interrupt();\n            throw new PulsarClientException(\"Close interrupted\", e);\n        } catch (ExecutionException e) {\n            throw new PulsarClientException(e.getCause());\n        }\n    }\n\n    // --- Async internals ---\n\n    CompletableFuture<Message<T>> receiveAsync() {\n        return receiveQueue.receiveAsync().thenApply(this::advanceCheckpoint);\n    }\n\n    CompletableFuture<Message<T>> receiveAsync(Duration timeout) {\n        return receiveQueue.receiveAsync(timeout).thenApply(this::advanceCheckpoint);\n    }\n\n    CompletableFuture<List<Message<T>>> receiveMultiAsync(int maxMessages, Duration timeout) {\n        return receiveQueue.receiveMultiAsync(maxMessages, timeout).thenApply(batch -> {\n            batch.forEach(this::advanceCheckpoint);","sourceCodeStart":205,"sourceCodeEnd":241,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java#L205-L241","documentation":"ScalableCheckpointConsumer.close() blocks on closeAsync().get(); if the waiting thread is interrupted, it restores the interrupt flag and throws a PulsarClientException with the fixed message 'Close interrupted'. This means the caller was interrupted while waiting for the consumer's async close to finish — the close itself may still be proceeding in the background.","triggerScenarios":"Calling ScalableCheckpointConsumer.close() from a thread that gets interrupted while blocked in closeAsync().get() — executor shutdownNow(), request-timeout interrupts, or application shutdown hooks.","commonSituations":"Checkpoint/recovery frameworks interrupting worker threads; container shutdown during a long-running close; cancellation frameworks interrupting tasks that own the consumer.","solutions":["Determine what interrupted the thread; the library already re-sets the interrupt flag, so honor it in your cleanup path","Avoid interrupting threads that own consumer close; prefer letting closeAsync() complete and attaching callbacks","If you must bound close time, use closeAsync() with orTimeout instead of interrupting the blocking get()"],"exampleFix":"// before\nconsumer.close(); // blocking, interrupt-sensitive\n// after\nconsumer.closeAsync().orTimeout(30, TimeUnit.SECONDS).join();","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    consumer.close();\n} catch (PulsarClientException e) {\n    if (Thread.currentThread().isInterrupted()) {\n        log.warn(\"Consumer close interrupted\");\n    } else {\n        throw e;\n    }\n}","preventionTips":["Don't interrupt threads that own consumer close; use closeAsync() with orTimeout for deadlines","Restore the interrupt flag in your own catch blocks","Close consumers in a controlled teardown step, not inside interruptible request handling"],"tags":["consumer","close","interrupted"],"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"}