{"record":{"id":"8238087d270edaa6","repo":"apache/pulsar","slug":"failed-to-create-producer-for-topic-topicname-p","errorCode":null,"errorMessage":"Failed to create Producer for topic ${topicName} producerName ${producerName} schema ${schemaToUse}","messagePattern":"Failed to create Producer for topic (.+?) producerName (.+?) schema (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"critical","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java","lineNumber":330,"sourceCode":"    public void close() throws Exception {\n        if (this.pulsarSinkProcessor != null) {\n            this.pulsarSinkProcessor.close();\n        }\n    }\n\n    Producer<T> createProducer(String topicName, Schema<T> schema, String producerName) {\n        Schema<T> schemaToUse = schema != null ? schema : this.schema;\n        try {\n            log.info()\n                    .attr(\"producerName\", producerName)\n                    .attr(\"topic\", topicName)\n                    .attr(\"schema\", schemaToUse)\n                    .log(\"Initializing producer\");\n            return producerBuilderFactory.createProducerBuilder(topicName, schemaToUse, producerName)\n                    .properties(properties)\n                    .create();\n        } catch (PulsarClientException e) {\n            throw new RuntimeException(\"Failed to create Producer for topic \" + topicName\n                    + \" producerName \" + producerName + \" schema \" + schemaToUse, e);\n        }\n    }\n\n    @SuppressWarnings(\"unchecked\")\n    @VisibleForTesting\n    Schema<T> initializeSchema() throws ClassNotFoundException {\n        if (StringUtils.isEmpty(this.pulsarSinkConfig.getTypeClassName())) {\n            return (Schema<T>) Schema.BYTES;\n        }\n\n        Class<?> typeArg = Reflections.loadClass(this.pulsarSinkConfig.getTypeClassName(), functionClassLoader);\n        if (Void.class.equals(typeArg)) {\n            // return type is 'void', so there's no schema to check\n            return null;\n        }\n        ConsumerConfig consumerConfig = new ConsumerConfig();\n        consumerConfig.setSchemaProperties(pulsarSinkConfig.getSchemaProperties());","sourceCodeStart":312,"sourceCodeEnd":348,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java#L312-L348","documentation":"PulsarSink.createProducer wraps PulsarClientException from building the sink's output producer in a RuntimeException naming the topic, producer name, and schema. It means the sink could not attach to the output topic (authorization, topic doesn't exist, client shutdown, schema incompatibility).","triggerScenarios":"producerBuilderFactory.createProducerBuilder(...).create() throwing PulsarClientException (connection failure, topic not found, NotAllowedException/authorization failure, schema serialization incompatibility).","commonSituations":"Output topic deleted or auto-creation disabled; function role lacks produce permissions on the topic; broker unreachable from the instance; schema set on sink conflicts with topic's existing schema.","solutions":["Check the wrapped PulsarClientException in the stack trace for the concrete cause.","Confirm the output topic exists (or enable topic auto-creation) and the function's role has produce permission (pulsar-admin topics grant-permission).","Verify broker connectivity (serviceUrl, network, TLS) from the function instance.","If schema-related, align the sink schema with the topic's schema or delete the conflicting topic schema."],"exampleFix":"// grant produce permission so createProducer succeeds\n// before: AuthorizationException wrapped in RuntimeException\n// after\npulsar-admin topics grant-permission -r produce persistent://tenant/ns/output-topic --role functions-role","handlingStrategy":"retry","validationCode":"// preflight\npulsar-admin topics permissions persistent://tenant/ns/output-topic\npulsar-admin topics lookup persistent://tenant/ns/output-topic","typeGuard":null,"tryCatchPattern":"try {\n  producer = createProducer(topic, schema, name);\n} catch (RuntimeException e) {\n  Throwable cause = e.getCause();\n  if (cause instanceof PulsarClientException) {\n    log.error(\"producer create failed for {}: {}\", topic, cause.getMessage());\n    // retry with backoff for transient causes, fail fast on authorization\n  }\n}","preventionTips":["Pre-create output topics and grant produce permission to the function role","Keep broker serviceUrl/TLS config correct in function instances","Verify schema compatibility before deploying a sink with a typed schema"],"tags":["producer","pulsar-client","authorization","pulsar-functions"],"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"}