What causes "failed to create producer: %w" in apache/pulsar?

Check the logged inner error for the root cause.

failed to create producer: %w

Error message

failed to create producer: %w

What it means

setupProducer creates the output producer for the function's sink topic. If getProducer fails (e.g. the Pulsar client cannot create the producer), the underlying error is logged and returned wrapped as "failed to create producer: %w", failing instance startup.

Source

Thrown at pulsar-function-go/pf/instance.go:250

	}

	client, err := pulsar.NewClient(clientOpts)
	if err != nil {
		log.Errorf("create client error:%v", err)
		gi.stats.incrTotalSysExceptions(err)
		return err
	}
	gi.client = client
	return nil
}

func (gi *goInstance) setupProducer() error {
	if gi.context.instanceConf.funcDetails.Sink.Topic != "" && len(gi.context.instanceConf.funcDetails.Sink.Topic) > 0 {
		log.Debugf("Setting up producer for topic %s", gi.context.instanceConf.funcDetails.Sink.Topic)
		producer, err := gi.getProducer(gi.context.instanceConf.funcDetails.Sink.Topic)
		if err != nil {
			log.Errorf("Failed to create producer: %v", err)
			return fmt.Errorf("failed to create producer: %w", err)
		}

		gi.producer = producer
	}
	return nil
}

func (gi *goInstance) getProducer(topicName string) (pulsar.Producer, error) {
	properties := getProperties(getDefaultSubscriptionName(
		gi.context.instanceConf.funcDetails.Tenant,
		gi.context.instanceConf.funcDetails.Namespace,
		gi.context.instanceConf.funcDetails.Name), gi.context.instanceConf.instanceID)

	// Compression and batching come from the function's producerSpec; everything else is fixed by
	// the runtime.
	options := producerOptionsFromSpec(gi.context.instanceConf.funcDetails.Sink.ProducerSpec)
	options.Topic = topicName
	options.Properties = properties

View on GitHub (pinned to 820761864e)

Solutions

Check the logged inner error for the root cause.
Verify brokerServiceURL connectivity from the function pod (telnet/curl the Pulsar port).
Confirm the sink topic exists or auto-creation is enabled, and the role has produce permission.
Validate the topic name format (persistent://tenant/namespace/topic).

Example fix

// before
sink:
  topic: persistent://public/default/my-sink   # broker unreachable
// after
# fix broker URL / network, then
sink:
  topic: persistent://public/default/my-sink

Defensive patterns

Strategy: retry

Validate before calling

// pre-check connectivity and topic before starting
conn, err := net.DialTimeout("tcp", brokerHostPort, 5*time.Second)
if err != nil {
    return fmt.Errorf("broker unreachable: %w", err)
}
conn.Close()

Try / catch

err := gi.setupProducer()
if err != nil {
    var pErr *pulsar.Error
    if errors.As(err, &pErr) && isRetryable(pErr) {
        time.Sleep(backoff)
        return gi.setupProducer()
    }
    return err
}

Prevention

Verify brokerServiceURL, DNS, and network policies before deploy.
Ensure the function role has produce permission on the sink topic.
Pre-create the sink topic or enable topic auto-creation.
Use valid persistent://tenant/namespace/topic names.

When it happens

Trigger: Sink topic configured but pulsar.NewProducer fails — broker unreachable, topic doesn't exist and auto-creation disabled, auth/authorization failure, or invalid topic name.

Common situations: Wrong broker service URL; namespace/topic lacks produce permission for the function's role; cluster down; DNS/network issues in k8s; invalid characters in topic name.

Related errors

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