apache/pulsar · error · RuntimeException

Unable to create producer for topic '%s'

Error message

Unable to create producer for topic '%s'

What it means

Inside getOrCreateProducer, the cache loader wraps checked exceptions from the producer-supplier Callable in a RuntimeException with this message; the original exception is the cause. It means Pulsar client's newProducer call failed while lazily creating a cached producer for the given topic. Only checked (non-Runtime) exceptions are wrapped — runtime failures propagate unchanged.

Source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java:143

        }
        cache = builder.build();
        CacheMetricsCollector.CAFFEINE.addCache("function-producer-cache", cache);
    }

    public <T> Producer<T> getOrCreateProducer(CacheArea cacheArea, String topicName, Object additionalCacheKey,
                                               Callable<Producer<T>> supplier) {
        if (closed.get()) {
            throw new IllegalStateException("ProducerCache is already closed");
        }
        @SuppressWarnings("unchecked")
        Producer<T> producer = (Producer<T>) cache.get(
                new ProducerCacheKey(cacheArea, topicName, additionalCacheKey), key -> {
            try {
                return supplier.call();
            } catch (RuntimeException e) {
                throw e;
            } catch (Exception e) {
                throw new RuntimeException("Unable to create producer for topic '" + topicName + "'", e);
            }
        });
        return producer;
    }

    public void close() {
        if (closed.compareAndSet(false, true)) {
            cache.invalidateAll();
            // schedule the waiting job on the cache executor
            cacheExecutor.execute(() -> {
                try {
                    FutureUtil.waitForAll(closeFutures).get();
                } catch (InterruptedException | ExecutionException e) {
                    log.warn().exception(e).log("Failed to close producers");
                }
            });
            // Wait for the cache executor to terminate.
            // The eviction jobs and waiting for the close futures to complete will run on the single-threaded

View on GitHub (pinned to 820761864e)

Solutions

  1. Inspect the cause (e.getCause()) for the underlying PulsarClientException to identify broker/auth/topic issues.
  2. Verify the topic exists or enable allowAutoTopicCreation on the broker.
  3. Grant the function's role produce permissions on the topic (namespace-level topic-level auth).
  4. Check broker connectivity (serviceUrl, network, TLS config) from the function instance.

Example fix

// before
Producer<byte[]> p = cache.getOrCreateProducer(area, "my-topic", null,
    () -> client.newProducer().topic("my-topic").create());
// after
try {
    admin.topics().createSubscriptionlessTopic("persistent://tenant/ns/my-topic");
} catch (PulsarAdminException.TopicAlreadyExistsException ignored) {
}
Producer<byte[]> p = cache.getOrCreateProducer(area, "my-topic", null,
    () -> client.newProducer().topic("my-topic").create());
Defensive patterns

Strategy: try-catch

Validate before calling

// before producing
boolean exists = admin.topics().getList(namespace).stream()
    .anyMatch(t -> t.endsWith(topicName));
boolean authorized = admin.namespaces().getPermissions(namespace)
    .getOrDefault(myRole, Collections.emptySet()).contains(AuthAction.produce);

Try / catch

try {
    Producer<T> p = cache.getOrCreateProducer(area, topic, key, supplier);
} catch (RuntimeException e) {
    Throwable cause = e.getCause();
    if (cause instanceof PulsarClientException) {
        // inspect for TopicNotFound / AuthorizationException / ConnectException
    }
    throw e;
}

Prevention

When it happens

Trigger: The supplier passed to getOrCreateProducer (typically client.newProducer().topic(topicName).create()) throws a checked exception — e.g. PulsarClientException from broker unavailability, authorization failure, or nonexistent topic — and cache.get() loads the key for the first time.

Common situations: Topic doesn't exist and auto-creation is disabled; missing producer permissions for the function's role; broker unreachable/DNS failure; PulsarClient already closed.

Related errors

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