{"record":{"id":"14b4ae4d893a0c10","repo":"apache/pulsar","slug":"wraps-underlying-failure-cause","errorCode":null,"errorMessage":"(wraps underlying failure cause)","messagePattern":"\\(wraps underlying failure cause\\)","errorType":"exception","errorClass":"PulsarClientException","httpStatus":null,"severity":"error","filePath":"pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java","lineNumber":388,"sourceCode":"        }\n        pendingDrainAcks.computeIfAbsent(segmentId, __ -> new ConcurrentLinkedQueue<>())\n                .add(ackFuture.exceptionally(ex -> null));\n    }\n\n    @Override\n    public AsyncStreamConsumer<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();\n    }\n\n    CompletableFuture<Message<T>> receiveAsync(Duration timeout) {\n        return receiveQueue.receiveAsync(timeout);\n    }\n\n    CompletableFuture<List<Message<T>>> receiveMultiAsync(int maxNumMessages, Duration timeout) {\n        return receiveQueue.receiveMultiAsync(maxNumMessages, timeout);\n    }\n\n    CompletableFuture<Void> closeAsync() {","sourceCodeStart":370,"sourceCodeEnd":406,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java#L370-L406","documentation":"close() converts a failed closeAsync() future into `new PulsarClientException(e.getCause())`; the thrown message is the underlying cause's message. For a stream consumer this typically means one of the segment consumers failed to close (broker error, connection loss, AlreadyClosed).","triggerScenarios":"closeAsync() completing exceptionally while closing the segment DAG: broker rejects close, network disconnects mid-close, or a segment consumer was already closed/failed.","commonSituations":"Application shutdown during broker failover; closing a stream consumer whose subscription was concurrently deleted; repeated close calls on a partially closed consumer.","solutions":["Inspect e.getCause() for the concrete broker/client error","Retry close after transient connectivity issues; the close is idempotent for already-closed segments","Check broker availability before shutdown; sequence shutdown: consumers → producers → client","Prefer closeAsync() and log each segment's failure with handle((c, ex) -> ...)"],"exampleFix":"// before\nstreamConsumer.close();\n// after\ntry {\n    streamConsumer.close();\n} catch (PulsarClientException e) {\n    log.error(\"stream close failed: {}\", e.getCause(), e);\n    // retry or continue shutdown depending on cause\n}","handlingStrategy":"try-catch","validationCode":"// verify broker connectivity before shutdown\n// client.getPartitionsForTopic(...) or a lightweight lookup as health check","typeGuard":null,"tryCatchPattern":"try {\n    streamConsumer.close();\n} catch (PulsarClientException e) {\n    Throwable cause = e.getCause();\n    log.error(\"stream close failed: {}\", cause, cause);\n    if (isTransient(cause)) streamConsumer.closeAsync();\n}","preventionTips":["Inspect the cause chain, not just the wrapper message","Retry close after transient broker/network failures","Sequence shutdown: consumers → producers → client","Use closeAsync() with per-segment error logging"],"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"}