{"record":{"id":"17f6ba974e0046e5","repo":"apache/pulsar","slug":"unable-to-create-producer-for-topic-s","errorCode":null,"errorMessage":"Unable to create producer for topic '%s'","messagePattern":"Unable to create producer for topic '(.+?)'","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java","lineNumber":143,"sourceCode":"        }\n        cache = builder.build();\n        CacheMetricsCollector.CAFFEINE.addCache(\"function-producer-cache\", cache);\n    }\n\n    public <T> Producer<T> getOrCreateProducer(CacheArea cacheArea, String topicName, Object additionalCacheKey,\n                                               Callable<Producer<T>> supplier) {\n        if (closed.get()) {\n            throw new IllegalStateException(\"ProducerCache is already closed\");\n        }\n        @SuppressWarnings(\"unchecked\")\n        Producer<T> producer = (Producer<T>) cache.get(\n                new ProducerCacheKey(cacheArea, topicName, additionalCacheKey), key -> {\n            try {\n                return supplier.call();\n            } catch (RuntimeException e) {\n                throw e;\n            } catch (Exception e) {\n                throw new RuntimeException(\"Unable to create producer for topic '\" + topicName + \"'\", e);\n            }\n        });\n        return producer;\n    }\n\n    public void close() {\n        if (closed.compareAndSet(false, true)) {\n            cache.invalidateAll();\n            // schedule the waiting job on the cache executor\n            cacheExecutor.execute(() -> {\n                try {\n                    FutureUtil.waitForAll(closeFutures).get();\n                } catch (InterruptedException | ExecutionException e) {\n                    log.warn().exception(e).log(\"Failed to close producers\");\n                }\n            });\n            // Wait for the cache executor to terminate.\n            // The eviction jobs and waiting for the close futures to complete will run on the single-threaded","sourceCodeStart":125,"sourceCodeEnd":161,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java#L125-L161","documentation":"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.","triggerScenarios":"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.","commonSituations":"Topic doesn't exist and auto-creation is disabled; missing producer permissions for the function's role; broker unreachable/DNS failure; PulsarClient already closed.","solutions":["Inspect the cause (e.getCause()) for the underlying PulsarClientException to identify broker/auth/topic issues.","Verify the topic exists or enable allowAutoTopicCreation on the broker.","Grant the function's role produce permissions on the topic (namespace-level topic-level auth).","Check broker connectivity (serviceUrl, network, TLS config) from the function instance."],"exampleFix":"// before\nProducer<byte[]> p = cache.getOrCreateProducer(area, \"my-topic\", null,\n    () -> client.newProducer().topic(\"my-topic\").create());\n// after\ntry {\n    admin.topics().createSubscriptionlessTopic(\"persistent://tenant/ns/my-topic\");\n} catch (PulsarAdminException.TopicAlreadyExistsException ignored) {\n}\nProducer<byte[]> p = cache.getOrCreateProducer(area, \"my-topic\", null,\n    () -> client.newProducer().topic(\"my-topic\").create());","handlingStrategy":"try-catch","validationCode":"// before producing\nboolean exists = admin.topics().getList(namespace).stream()\n    .anyMatch(t -> t.endsWith(topicName));\nboolean authorized = admin.namespaces().getPermissions(namespace)\n    .getOrDefault(myRole, Collections.emptySet()).contains(AuthAction.produce);","typeGuard":null,"tryCatchPattern":"try {\n    Producer<T> p = cache.getOrCreateProducer(area, topic, key, supplier);\n} catch (RuntimeException e) {\n    Throwable cause = e.getCause();\n    if (cause instanceof PulsarClientException) {\n        // inspect for TopicNotFound / AuthorizationException / ConnectException\n    }\n    throw e;\n}","preventionTips":["Enable allowAutoTopicCreation or pre-create the function's output topics.","Grant produce permission to the function's role before deploying.","Verify serviceUrl/TLS settings from the instance's network position.","Ensure the PulsarClient outlives the cache and is not closed first."],"tags":["producer","pulsar-client","topic","runtime-exception"],"backgroundTag":"producer-creation-failed","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}