{"record":{"id":"9a5b73265af3b31c","repo":"apache/beam","slug":"failed-to-connect-to-control-service","errorCode":null,"errorMessage":"failed to connect to control service","messagePattern":"failed to connect to control service","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"sdks/go/pkg/beam/core/runtime/harness/harness.go","lineNumber":122,"sourceCode":"\t\tlog.Debugf(ctx, \"Failed to parse element_processing_timeout: %v, there will be no timeout for processing an element in a PTransform operation\", err)\n\t}\n\n\t// Connect to FnAPI control server. Receive and execute work.\n\tconn, err := dial(ctx, controlEndpoint, \"control\", 60*time.Second)\n\tif err != nil {\n\t\treturn errors.Wrap(err, \"failed to connect\")\n\t}\n\tdefer conn.Close()\n\n\tclient := fnpb.NewBeamFnControlClient(conn)\n\n\tlookupDesc := func(id bundleDescriptorID) (*fnpb.ProcessBundleDescriptor, error) {\n\t\treturn client.GetProcessBundleDescriptor(ctx, &fnpb.GetProcessBundleDescriptorRequest{ProcessBundleDescriptorId: string(id)})\n\t}\n\n\tstub, err := client.Control(ctx)\n\tif err != nil {\n\t\treturn errors.Wrapf(err, \"failed to connect to control service\")\n\t}\n\n\tlog.Debugf(ctx, \"Successfully connected to control @ %v\", controlEndpoint)\n\n\t// Each ProcessBundle is a sub-graph of the original one.\n\n\tvar wg sync.WaitGroup\n\trespc := make(chan *fnpb.InstructionResponse, 100)\n\n\twg.Add(1)\n\n\t// gRPC requires all writers to a stream be the same goroutine, so this is the\n\t// goroutine for managing responses back to the control service.\n\tgo func() {\n\t\tdefer wg.Done()\n\t\tfor resp := range respc {\n\t\t\t// TODO(lostluck): 2023/03/29 fix debug level logging to be flagged.\n\t\t\t// log.Debugf(ctx, \"RESP: %v\", proto.MarshalTextString(resp))","sourceCodeStart":104,"sourceCodeEnd":140,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/harness/harness.go#L104-L140","documentation":"After dialing the control server, the harness opens the bidirectional gRPC Control stream via client.Control(ctx). If the stream RPC itself fails (server unavailable, RPC unimplemented, deadline exceeded, stream refused), the error is wrapped as \"failed to connect to control service\" and MainWithOptions returns. Unlike a dial failure, the TCP/TLS connection existed but the Control RPC could not be established.","triggerScenarios":"client.Control(ctx) returns an error: the control server does not expose/accept the Control stream, the context deadline expires, the server rejects the RPC (unimplemented/permission denied), or the connection drops during stream setup.","commonSituations":"Runner and SDK version mismatch causing an incompatible BeamFnControl service; control server crashed between dial and Control; per-RPC deadline exceeded under load; mTLS/auth misconfiguration on the stream.","solutions":["Check the wrapped gRPC status code: Unimplemented -> version mismatch, DeadlineExceeded -> raise timeout/resources, Unavailable -> retry or fix startup ordering.","Align Beam runner and Go SDK harness versions so the BeamFnControl service contract matches.","Confirm the control server process is healthy and not crash-looping at the moment of stream setup.","Retry the worker; transient Unavailable during startup is common in container environments.","Inspect runner logs alongside harness logs for the server-side reason the stream was refused."],"exampleFix":"// before: no retry around stream setup\nstub, err := client.Control(ctx)\nif err != nil {\n    return errors.Wrapf(err, \"failed to connect to control service\")\n}\n// after: tolerate transient unavailability at startup\nvar stub fnpb.BeamFnControl_ControlClient\nbackoff := 1 * time.Second\nfor attempt := 0; attempt < 5; attempt++ {\n    stub, err = client.Control(ctx)\n    if err == nil {\n        break\n    }\n    if status.Code(err) == codes.Unimplemented {\n        return errors.Wrapf(err, \"failed to connect to control service\")\n    }\n    time.Sleep(backoff)\n    backoff *= 2\n}\nif err != nil {\n    return errors.Wrapf(err, \"failed to connect to control service\")\n}","handlingStrategy":"retry","validationCode":"if controlEndpoint == \"\" {\n    return errors.New(\"control endpoint must be set before opening the Control stream\")\n}\n// dial with grpc.WithBlock() + grpc.WaitForReady(true) so the channel is\n// confirmed ready before attempting to open the Control stream","typeGuard":null,"tryCatchPattern":"stub, err := client.Control(ctx)\nif err != nil {\n    switch status.Code(err) {\n    case codes.Unimplemented:\n        // version mismatch between runner and SDK: fix versions, do not retry\n    case codes.Unavailable, codes.DeadlineExceeded:\n        // transient: retry with backoff\n    default:\n        return errors.Wrapf(err, \"failed to connect to control service\")\n    }\n}","preventionTips":["Keep Beam runner and Go SDK container versions in lockstep.","Add gRPC keepalive options so stream setup survives constrained networks.","Check server logs for stream rejection reasons when this fires reproducibly.","Retry transient failures; never retry Unimplemented."],"tags":["network","grpc","beam-go","rpc","startup"],"backgroundTag":"grpc-stream-setup-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}