apache/beam · error

failed to connect to state service %v

Error message

failed to connect to state service %v

What it means

newStateChannel dials the Beam Fn State gRPC service at the port URL (15s timeout). If the dial fails, it wraps the error with the state service URL. This is a connectivity failure between the SDK harness and the runner's state service (or a missing/incorrect state ApiServiceDescriptor).

Source

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

func (c *StateChannel) terminateStreamOnError(err error) {
	c.mu.Lock()
	if c.forceRecreate != nil {
		c.closedErr = err
		c.forceRecreate(c.id, err)
		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),

View on GitHub (pinned to 12126d8942)

Solutions

1. Verify network connectivity from the worker to the state service endpoint URL and that no proxy/firewall blocks gRPC.
2. Confirm the runner is healthy and still serving the state port; check runner logs for crashes.
3. Check TLS/certificate configuration for the gRPC endpoint.
4. Retry the pipeline; if reproducible on DirectRunner, file a Beam issue with the full wrapped error.

conn, err := net.DialTimeout("tcp", host, 5*time.Second)
if err != nil { return fmt.Errorf("state service %s unreachable: %w", host, err) }
conn.Close()

ch, err := mgr.Open(ctx, port)
if err != nil && strings.Contains(err.Error(), "failed to connect to state service") {
    return retry.Do(func() error { _, err := mgr.Open(ctx, port); return err }, retry.Attempts(3))
}
return err

When it happens

Trigger: Calling Open on the state manager when the runner-provided state gRPC endpoint is unreachable, TLS misconfigured, the runner died, or in direct mode where no state service exists and a test executor sets up its own endpoints.

Common situations: Network partitions between harness worker and runner, wrong endpoint from runner config, firewall/proxy blocking gRPC, runner OOM/crash mid-bundle, using state APIs on a runner without state service support.

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

• failed to send chunks for %v; close error: %v
• chunk send failed
• StateChannel[%v].Send(%v): channel closed due to: %v
• failed to connect: %v
• StageFile chunk send failed

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