apache/pulsar · critical · RuntimeException

Failed to create Producer for topic

Error message

Failed to create Producer for topic ${topicName} producerName ${producerName} schema ${schemaToUse}

What it means

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).

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.

Example fix

// grant produce permission so createProducer succeeds
// before: AuthorizationException wrapped in RuntimeException
// after
pulsar-admin topics grant-permission -r produce persistent://tenant/ns/output-topic --role functions-role

Defensive patterns

Strategy: retry

Validate before calling

// preflight
pulsar-admin topics permissions persistent://tenant/ns/output-topic
pulsar-admin topics lookup persistent://tenant/ns/output-topic

Try / catch

try {
  producer = createProducer(topic, schema, name);
} catch (RuntimeException e) {
  Throwable cause = e.getCause();
  if (cause instanceof PulsarClientException) {
    log.error("producer create failed for {}: {}", topic, cause.getMessage());
    // retry with backoff for transient causes, fail fast on authorization
  }
}

Prevention

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

When it happens

Trigger: producerBuilderFactory.createProducerBuilder(...).create() throwing PulsarClientException (connection failure, topic not found, NotAllowedException/authorization failure, schema serialization incompatibility).

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

Related errors

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

Appendix: source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java:330

    public void close() throws Exception {
        if (this.pulsarSinkProcessor != null) {
            this.pulsarSinkProcessor.close();
        }
    }

    Producer<T> createProducer(String topicName, Schema<T> schema, String producerName) {
        Schema<T> schemaToUse = schema != null ? schema : this.schema;
        try {
            log.info()
                    .attr("producerName", producerName)
                    .attr("topic", topicName)
                    .attr("schema", schemaToUse)
                    .log("Initializing producer");
            return producerBuilderFactory.createProducerBuilder(topicName, schemaToUse, producerName)
                    .properties(properties)
                    .create();
        } catch (PulsarClientException e) {
            throw new RuntimeException("Failed to create Producer for topic " + topicName
                    + " producerName " + producerName + " schema " + schemaToUse, e);
        }
    }

    @SuppressWarnings("unchecked")
    @VisibleForTesting
    Schema<T> initializeSchema() throws ClassNotFoundException {
        if (StringUtils.isEmpty(this.pulsarSinkConfig.getTypeClassName())) {
            return (Schema<T>) Schema.BYTES;
        }

        Class<?> typeArg = Reflections.loadClass(this.pulsarSinkConfig.getTypeClassName(), functionClassLoader);
        if (Void.class.equals(typeArg)) {
            // return type is 'void', so there's no schema to check
            return null;
        }
        ConsumerConfig consumerConfig = new ConsumerConfig();
        consumerConfig.setSchemaProperties(pulsarSinkConfig.getSchemaProperties());

View on GitHub (pinned to 820761864e)