apache/pulsar · critical

Failed to unmarshal consume specs: %v

Error message

Failed to unmarshal consume specs: %v

What it means

newInstanceConfWithConf decodes each entry of SourceInputSpecs (a map of topic -> JSON string) into a pb.ConsumerSpec. If any value is not valid JSON, the instance config construction panics with this message, crashing the function instance at startup.

Source

Thrown at pulsar-function-go/pf/instanceConf.go:71

	tlsHostnameVerification     bool
}

func newInstanceConfWithConf(cfg *conf.Conf) *instanceConf {
	inputSpecs := make(map[string]*pb.ConsumerSpec)
	// for backward compatibility
	if cfg.SourceSpecTopic != "" {
		inputSpecs[cfg.SourceSpecTopic] = &pb.ConsumerSpec{
			SchemaType:     cfg.SourceSchemaType,
			IsRegexPattern: cfg.IsRegexPatternSubscription,
			ReceiverQueueSize: &pb.ConsumerSpec_ReceiverQueueSize{
				Value: cfg.ReceiverQueueSize,
			},
		}
	}
	for topic, value := range cfg.SourceInputSpecs {
		spec := &pb.ConsumerSpec{}
		if err := json.Unmarshal([]byte(value), spec); err != nil {
			panic(fmt.Sprintf("Failed to unmarshal consume specs: %v", err))
		}
		inputSpecs[topic] = spec
	}
	instanceConf := &instanceConf{
		instanceID:                  cfg.InstanceID,
		funcID:                      cfg.FuncID,
		funcVersion:                 cfg.FuncVersion,
		maxBufTuples:                cfg.MaxBufTuples,
		port:                        cfg.Port,
		clusterName:                 cfg.ClusterName,
		pulsarServiceURL:            cfg.PulsarServiceURL,
		stateServiceURL:             cfg.StateStorageServiceURL,
		pulsarWebServiceURL:         cfg.PulsarWebServiceURL,
		killAfterIdle:               cfg.KillAfterIdleMs,
		expectedHealthCheckInterval: cfg.ExpectedHealthCheckInterval,
		metricsPort:                 cfg.MetricsPort,
		funcDetails: pb.FunctionDetails{
			Tenant:               cfg.Tenant,

View on GitHub (pinned to 820761864e)

Solutions

1. Validate each SourceInputSpecs value parses as a ConsumerSpec JSON object (e.g. {"schemaType":"...","receiverQueueSize":...}).
2. Fix the component that populates SourceInputSpecs (usually the function worker) to serialize specs properly.
3. Catch the panic in a wrapper during local testing to surface the offending topic/value.

Example fix

// before
"sourceInputSpecs": {"topic-a": ""} // panics on unmarshal
// after
"sourceInputSpecs": {"topic-a": "{\"receiverQueueSize\":0}"}

for topic, raw := range sourceInputSpecs {
    var spec pb.ConsumerSpec
    if err := json.Unmarshal([]byte(raw), &spec); err != nil {
        return fmt.Errorf("invalid ConsumerSpec JSON for topic %s: %w", topic, err)
    }
}

defer func() {
    if r := recover(); r != nil {
        if s, ok := r.(string); ok && strings.Contains(s, "unmarshal consume specs") {
            log.Fatalf("malformed SourceInputSpecs: %s", s)
        }
        panic(r)
    }
}()

When it happens

Trigger: A SourceInputSpecs value that isn't valid JSON for ConsumerSpec, e.g. an empty string, plain topic name, or truncated/invalid JSON produced by the function worker config serialization.

Common situations: Manual conf file edits introducing malformed JSON; upstream framework passing raw values instead of serialized ConsumerSpec; encoding bugs between worker and Go instance.

Understand the failure class

Background: "failed to unmarshal" / json.Unmarshal errors: why parsing a response into a Go struct fails and how to fix it — this error's family across 23 libraries.

Related errors

• Failed to parse tlsFactoryConfig as a JSON object
• Malformed authentication parameters
• Failed to load config into existing configuration data
• Failed to parse SASL authParams
• Cannot parse sink-config

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