apache/pulsar · error · PulsarClientException

(wraps subscribeAsync failure cause)

Error message

(wraps subscribeAsync failure cause)

What it means

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.

Source

Thrown at pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/StreamConsumerBuilderV5.java:62

    // single-topic vs multi-topic mode.
    private String topicName;
    private org.apache.pulsar.common.naming.NamespaceName namespaceName;
    private Map<String, String> propertyFilters;

    StreamConsumerBuilderV5(PulsarClientV5 client, Schema<T> v5Schema) {
        this.client = client;
        this.v5Schema = v5Schema;
    }

    @Override
    public StreamConsumer<T> subscribe() throws PulsarClientException {
        try {
            return subscribeAsync().join();
        } catch (java.util.concurrent.CompletionException e) {
            if (e.getCause() instanceof PulsarClientException pce) {
                throw pce;
            }
            throw new PulsarClientException(e.getCause());
        }
    }

    @Override
    public CompletableFuture<StreamConsumer<T>> subscribeAsync() {
        boolean topicSet = topicName != null && !topicName.isEmpty();
        boolean namespaceSet = namespaceName != null;
        if (topicSet == namespaceSet) {
            return CompletableFuture.failedFuture(
                    new PulsarClientException.InvalidConfigurationException(
                            "Exactly one of .topic(name) or .namespace(...) must be set"));
        }
        if (conf.getSubscriptionName() == null || conf.getSubscriptionName().isEmpty()) {
            return CompletableFuture.failedFuture(
                    new PulsarClientException.InvalidConfigurationException("Subscription name is required"));
        }
        // Default the consumer name to a stable random when the user didn't set one —
        // ScalableConsumerClient uses it as the registration key with the controller.

View on GitHub (pinned to 820761864e)

Solutions

1. Inspect the cause chain of the thrown PulsarClientException (getCause()) to find the actual subscription failure.
2. Validate topicName and serviceUrl on the builder before calling subscribe().
3. Confirm the broker is reachable and credentials/schema config are correct; test with subscribeAsync() to get the failure without blocking.
4. If a non-Pulsar exception is wrapped, fix the underlying bug (null config, bad schema class) indicated by the cause.

Example fix

// before
StreamConsumer<String> consumer = new StreamConsumerBuilderV5<String>()
    .topic(topicName)
    .subscribe();
// after
if (topicName == null || topicName.isEmpty()) {
    throw new IllegalArgumentException("topic must be set before subscribe()");
}
StreamConsumer<String> consumer;
try {
    consumer = new StreamConsumerBuilderV5<String>()
        .topic(topicName)
        .subscribe();
} catch (PulsarClientException e) {
    log.error("subscribe failed: {}", e.getCause(), e);
    throw e;
}

if (topicName == null || topicName.isEmpty()) throw new IllegalArgumentException("topic must be set");
if (serviceUrl == null || serviceUrl.isEmpty()) throw new IllegalArgumentException("serviceUrl must be set");

static boolean isPulsarFailure(Throwable t) {
    return t instanceof PulsarClientException || (t.getCause() instanceof PulsarClientException);
}

try {
    StreamConsumer<T> c = builder.subscribe();
} catch (PulsarClientException e) {
    Throwable root = e.getCause() != null ? e.getCause() : e;
    log.error("subscribe failed: {}", root.getMessage(), root);
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• (wraps underlying failure cause)
• (wraps v4 producer error message)
• (wraps commit failure cause message)
• (wraps abort failure cause message)
• Failed to build TLS instanceClass for purpose purpose

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/d8e09be01ac0741c. Report an issue: GitHub.