{"record":{"id":"77e1e1a5906e6316","repo":"apache/pulsar","slug":"cause-77e1e1","errorCode":null,"errorMessage":"${cause}","messagePattern":"\\$\\{cause\\}","errorType":"exception","errorClass":"org.apache.pulsar.client.impl.v5.PulsarClientException","httpStatus":null,"severity":"error","filePath":"pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java","lineNumber":225,"sourceCode":"    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);\n            return batch;\n        });","sourceCodeStart":207,"sourceCodeEnd":243,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java#L207-L243","documentation":"ScalableCheckpointConsumer.close() blocks on closeAsync().get() and wraps any failure of the asynchronous close in a PulsarClientException. When the future completes exceptionally, ExecutionException is caught and the underlying cause is rethrown as `new PulsarClientException(e.getCause())`, so `${cause}` is the message of the real underlying failure (e.g. broker disconnect, already-closed consumer).","triggerScenarios":"Calling close() when the underlying async close future completes exceptionally — e.g. a segment consumer failed to close, the broker rejected the close (ConsumerBusy/AlreadyClosed), or a network error occurred while unsubscribing/checkpoint state was being cleaned up.","commonSituations":"Shutting down an application while the broker is unreachable; double-closing a consumer; closing during a broker restart; timeouts while closing many segments of a scalable checkpoint consumer.","solutions":["Inspect the cause chain of the thrown PulsarClientException (getCause()) to find the real close failure","Verify broker connectivity and that the broker is up before/while closing","Avoid closing the same consumer twice; guard with an isClosed flag or try-finally","Retry close() after a transient network failure; consider closeAsync() for finer control"],"exampleFix":"// before\nconsumer.close();\n// after\ntry {\n    consumer.close();\n} catch (PulsarClientException e) {\n    log.warn(\"close failed: {}\", e.getCause(), e);\n    // inspect e.getCause() for the underlying failure and retry if transient\n}","handlingStrategy":"try-catch","validationCode":"// no pre-check API; ensure client is connected and consumer not already closed\nif (consumer == null || alreadyClosed) { skipClose(); }","typeGuard":null,"tryCatchPattern":"try {\n    consumer.close();\n} catch (PulsarClientException e) {\n    Throwable cause = e.getCause();\n    log.warn(\"close failed: {}\", cause, cause);\n    if (isTransient(cause)) retryClose(consumer);\n}","preventionTips":["Close consumers in a shutdown hook before stopping executors","Never close the same consumer twice","Check broker health before shutdown","Use closeAsync() when you need per-failure control"],"tags":["pulsar","close","lifecycle","network"],"backgroundTag":"consumer-close-failed","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"}