apache/beam · error

failed to create state client

Error message

failed to create state client %v

What it means

After a successful dial, newStateChannel opens the BeamFnState bidirectional stream via fnpb.NewBeamFnStateClient(cc).State(ctx). If the stream cannot be established, the connection is closed and this wrapped error is returned with the port URL. The endpoint accepted the TCP/gRPC handshake but the State RPC itself failed.

Solutions

1. Align Beam SDK and runner versions (protocol skew is the most common cause).
2. Check runner logs for errors accepting the State stream.
3. Inspect the inner error for context.DeadlineExceeded and retry the bundle/job if transient.
4. If persistent, report to the runner maintainer with the wrapped error and endpoint.

// ensure versions match
if runtimeVersion != runnerApiVersion { return fmt.Errorf("SDK %s vs runner %s mismatch", runtimeVersion, runnerApiVersion) }

if err != nil && strings.Contains(err.Error(), "failed to create state client") {
    time.Sleep(time.Second)
    return retryOpen(ctx, port) // transient stream-setup failure
}

When it happens

Trigger: The state service rejecting or failing the State() streaming RPC — e.g. server does not implement BeamFnState, protocol version mismatch, request canceled due to the 15s context deadline, or runner-side error at stream setup.

Common situations: Runner/SDK version skew where the state service API differs, runner closing connections during shutdown, resource exhaustion on the runner preventing new streams.

Related errors

• chunk send failed
• error creating local job server
• failed to close stream for
• failed to connect to state service
• failed to connect

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/7f8bc434d90edf97. Report an issue: GitHub.

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/harness/statemgr.go:678

		c.forceRecreate = nil
	}
	// Cancelling context after forcing recreation to ensure closedErr is set.
	c.cancelFn()
	c.mu.Unlock()
}

func newStateChannel(ctx context.Context, port exec.Port) (*StateChannel, error) {
	ctx, cancelFn := context.WithCancel(ctx)
	cc, err := dial(ctx, port.URL, "state", 15*time.Second)
	if err != nil {
		cancelFn()
		return nil, errors.Wrapf(err, "failed to connect to state service %v", port.URL)
	}
	client, err := fnpb.NewBeamFnStateClient(cc).State(ctx)
	if err != nil {
		cc.Close()
		cancelFn()
		return nil, errors.Wrapf(err, "failed to create state client %v", port.URL)
	}
	return makeStateChannel(ctx, port.URL, client, func() {
		cc.Close()
		cancelFn()
	}), nil
}

func makeStateChannel(ctx context.Context, id string, client stateClient, cancelFn context.CancelFunc) *StateChannel {
	ret := &StateChannel{
		id:        id,
		client:    client,
		requests:  make(chan *fnpb.StateRequest, 10),
		responses: make(map[string]chan<- *fnpb.StateResponse),
		cancelFn:  cancelFn,
		DoneCh:    ctx.Done(),
	}
	go ret.read(ctx)
	go ret.write(ctx)

View on GitHub (pinned to 12126d8942)