{"record":{"id":"cf725b2646916e50","repo":"apache/pulsar","slug":"failed-to-create-producer-w","errorCode":null,"errorMessage":"failed to create producer: %w","messagePattern":"failed to create producer: %w","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"pulsar-function-go/pf/instance.go","lineNumber":250,"sourceCode":"\t}\n\n\tclient, err := pulsar.NewClient(clientOpts)\n\tif err != nil {\n\t\tlog.Errorf(\"create client error:%v\", err)\n\t\tgi.stats.incrTotalSysExceptions(err)\n\t\treturn err\n\t}\n\tgi.client = client\n\treturn nil\n}\n\nfunc (gi *goInstance) setupProducer() error {\n\tif gi.context.instanceConf.funcDetails.Sink.Topic != \"\" && len(gi.context.instanceConf.funcDetails.Sink.Topic) > 0 {\n\t\tlog.Debugf(\"Setting up producer for topic %s\", gi.context.instanceConf.funcDetails.Sink.Topic)\n\t\tproducer, err := gi.getProducer(gi.context.instanceConf.funcDetails.Sink.Topic)\n\t\tif err != nil {\n\t\t\tlog.Errorf(\"Failed to create producer: %v\", err)\n\t\t\treturn fmt.Errorf(\"failed to create producer: %w\", err)\n\t\t}\n\n\t\tgi.producer = producer\n\t}\n\treturn nil\n}\n\nfunc (gi *goInstance) getProducer(topicName string) (pulsar.Producer, error) {\n\tproperties := getProperties(getDefaultSubscriptionName(\n\t\tgi.context.instanceConf.funcDetails.Tenant,\n\t\tgi.context.instanceConf.funcDetails.Namespace,\n\t\tgi.context.instanceConf.funcDetails.Name), gi.context.instanceConf.instanceID)\n\n\t// Compression and batching come from the function's producerSpec; everything else is fixed by\n\t// the runtime.\n\toptions := producerOptionsFromSpec(gi.context.instanceConf.funcDetails.Sink.ProducerSpec)\n\toptions.Topic = topicName\n\toptions.Properties = properties","sourceCodeStart":232,"sourceCodeEnd":268,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-function-go/pf/instance.go#L232-L268","documentation":"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.","triggerScenarios":"Sink topic configured but pulsar.NewProducer fails — broker unreachable, topic doesn't exist and auto-creation disabled, auth/authorization failure, or invalid topic name.","commonSituations":"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.","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)."],"exampleFix":"// before\nsink:\n  topic: persistent://public/default/my-sink   # broker unreachable\n// after\n# fix broker URL / network, then\nsink:\n  topic: persistent://public/default/my-sink","handlingStrategy":"retry","validationCode":"// pre-check connectivity and topic before starting\nconn, err := net.DialTimeout(\"tcp\", brokerHostPort, 5*time.Second)\nif err != nil {\n    return fmt.Errorf(\"broker unreachable: %w\", err)\n}\nconn.Close()","typeGuard":null,"tryCatchPattern":"err := gi.setupProducer()\nif err != nil {\n    var pErr *pulsar.Error\n    if errors.As(err, &pErr) && isRetryable(pErr) {\n        time.Sleep(backoff)\n        return gi.setupProducer()\n    }\n    return err\n}","preventionTips":["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."],"tags":["go","producer","network","pulsar-client"],"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"}