{"record":{"id":"d8e09be01ac0741c","repo":"apache/pulsar","slug":"wraps-subscribeasync-failure-cause","errorCode":null,"errorMessage":"(wraps subscribeAsync failure cause)","messagePattern":"\\(wraps subscribeAsync failure cause\\)","errorType":"exception","errorClass":"PulsarClientException","httpStatus":null,"severity":"error","filePath":"pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/StreamConsumerBuilderV5.java","lineNumber":62,"sourceCode":"    // single-topic vs multi-topic mode.\n    private String topicName;\n    private org.apache.pulsar.common.naming.NamespaceName namespaceName;\n    private Map<String, String> propertyFilters;\n\n    StreamConsumerBuilderV5(PulsarClientV5 client, Schema<T> v5Schema) {\n        this.client = client;\n        this.v5Schema = v5Schema;\n    }\n\n    @Override\n    public StreamConsumer<T> subscribe() throws PulsarClientException {\n        try {\n            return subscribeAsync().join();\n        } catch (java.util.concurrent.CompletionException e) {\n            if (e.getCause() instanceof PulsarClientException pce) {\n                throw pce;\n            }\n            throw new PulsarClientException(e.getCause());\n        }\n    }\n\n    @Override\n    public CompletableFuture<StreamConsumer<T>> subscribeAsync() {\n        boolean topicSet = topicName != null && !topicName.isEmpty();\n        boolean namespaceSet = namespaceName != null;\n        if (topicSet == namespaceSet) {\n            return CompletableFuture.failedFuture(\n                    new PulsarClientException.InvalidConfigurationException(\n                            \"Exactly one of .topic(name) or .namespace(...) must be set\"));\n        }\n        if (conf.getSubscriptionName() == null || conf.getSubscriptionName().isEmpty()) {\n            return CompletableFuture.failedFuture(\n                    new PulsarClientException.InvalidConfigurationException(\"Subscription name is required\"));\n        }\n        // Default the consumer name to a stable random when the user didn't set one —\n        // ScalableConsumerClient uses it as the registration key with the controller.","sourceCodeStart":44,"sourceCodeEnd":80,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/StreamConsumerBuilderV5.java#L44-L80","documentation":"StreamConsumerBuilderV5.subscribe() is a synchronous convenience wrapper that blocks on subscribeAsync().join(). When the async subscription completes exceptionally, the CompletionException is unwrapped: a PulsarClientException cause is rethrown directly, and any other cause is wrapped in a new PulsarClientException. The resulting message is derived from the underlying failure cause, so this error indicates the real reason the stream subscription failed.","triggerScenarios":"Calling subscribe() when the underlying subscribeAsync() future fails — e.g. invalid/empty topic name, broker unreachable, authentication failure, or a non-Pulsar exception (NPE, IllegalArgument) thrown inside the async pipeline.","commonSituations":"Typo or empty topic/serviceUrl configuration; broker down or TLS/auth misconfigured; schema incompatibility for type T; upgrading from the v4 client and passing options the v5 builder does not accept, causing an internal error.","solutions":["Inspect the cause chain of the thrown PulsarClientException (getCause()) to find the actual subscription failure.","Validate topicName and serviceUrl on the builder before calling subscribe().","Confirm the broker is reachable and credentials/schema config are correct; test with subscribeAsync() to get the failure without blocking.","If a non-Pulsar exception is wrapped, fix the underlying bug (null config, bad schema class) indicated by the cause."],"exampleFix":"// before\nStreamConsumer<String> consumer = new StreamConsumerBuilderV5<String>()\n    .topic(topicName)\n    .subscribe();\n// after\nif (topicName == null || topicName.isEmpty()) {\n    throw new IllegalArgumentException(\"topic must be set before subscribe()\");\n}\nStreamConsumer<String> consumer;\ntry {\n    consumer = new StreamConsumerBuilderV5<String>()\n        .topic(topicName)\n        .subscribe();\n} catch (PulsarClientException e) {\n    log.error(\"subscribe failed: {}\", e.getCause(), e);\n    throw e;\n}","handlingStrategy":"try-catch","validationCode":"if (topicName == null || topicName.isEmpty()) throw new IllegalArgumentException(\"topic must be set\");\nif (serviceUrl == null || serviceUrl.isEmpty()) throw new IllegalArgumentException(\"serviceUrl must be set\");","typeGuard":"static boolean isPulsarFailure(Throwable t) {\n    return t instanceof PulsarClientException || (t.getCause() instanceof PulsarClientException);\n}","tryCatchPattern":"try {\n    StreamConsumer<T> c = builder.subscribe();\n} catch (PulsarClientException e) {\n    Throwable root = e.getCause() != null ? e.getCause() : e;\n    log.error(\"subscribe failed: {}\", root.getMessage(), root);\n}","preventionTips":["Always set topic and serviceUrl before subscribe()","Prefer subscribeAsync() with whenComplete to see the raw cause without blocking","Verify broker connectivity and credentials with a simple producer/consumer smoke test","Check schema compatibility for the generic type T before subscribing"],"tags":["subscription","pulsar-client","async","wrapped-exception"],"backgroundTag":"subscribe-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"}