{"record":{"id":"573daf93d1f6a9dd","repo":"apache/beam","slug":"failed-to-connect-to-data-service-at-v","errorCode":null,"errorMessage":"failed to connect to data service at %v","messagePattern":"failed to connect to data service at (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"sdks/go/pkg/beam/core/runtime/harness/datamgr.go","lineNumber":262,"sourceCode":"// If permitted, PTransformDone closes the channel.\nfunc (ec *elementsChan) PTransformDone() {\n\tec.mu.Lock()\n\tdefer ec.mu.Unlock()\n\tec.got++\n\tif ec.want > 0 && ec.want == ec.got {\n\t\tif !ec.Closed() {\n\t\t\tatomic.StoreUint32(&ec.closed, 1)\n\t\t\tclose(ec.ch)\n\t\t}\n\t}\n}\n\nfunc newDataChannel(ctx context.Context, port exec.Port) (*DataChannel, error) {\n\tctx, cancelFn := context.WithCancel(ctx)\n\tcc, err := dial(ctx, port.URL, \"data\", 15*time.Second)\n\tif err != nil {\n\t\tcancelFn()\n\t\treturn nil, errors.Wrapf(err, \"failed to connect to data service at %v\", port.URL)\n\t}\n\tclient, err := fnpb.NewBeamFnDataClient(cc).Data(ctx)\n\tif err != nil {\n\t\tcc.Close()\n\t\tcancelFn()\n\t\treturn nil, errors.Wrapf(err, \"failed to create data client on %v\", port.URL)\n\t}\n\treturn makeDataChannel(ctx, port.URL, client, func() {\n\t\tcc.Close()\n\t\tcancelFn()\n\t}), nil\n}\n\nfunc makeDataChannel(ctx context.Context, id string, client dataClient, cancelFn context.CancelFunc) *DataChannel {\n\tret := &DataChannel{\n\t\tid:                id,\n\t\tclient:            client,\n\t\twriters:           make(map[instructionID]map[string]*dataWriter),","sourceCodeStart":244,"sourceCodeEnd":280,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/harness/datamgr.go#L244-L280","documentation":"newDataChannel failed to establish a gRPC connection to the Beam data service at the URL carried in the execution port, within a 15 second dial timeout. This is a transport-level failure: the logging/data gRPC endpoint was unreachable or refused the connection. The library wraps the underlying dial error with the target URL for diagnosis.","triggerScenarios":"The data service host:port in port.URL is unreachable; the FnHarness was started with a stale or wrong artifact/log service endpoint; networking between SDK worker container and the runner's data service is blocked; the data service is not yet listening when the worker dials (15s timeout exceeded).","commonSituations":"Running Beam Go on Flink/Spark/Dataflow with misconfigured worker networking or missing ingress rules; docker-network isolation where the worker cannot resolve the runner hostname; Kubernetes NetworkPolicy blocking worker-to-runner traffic; local DirectRunner port conflicts.","solutions":["Verify the port.URL host/port is reachable from the worker (nc/host dig) and that the data service is up","Fix worker networking: run containers on a shared network, open firewall/NetworkPolicy rules","Increase or inspect the 15s dial window; if the service starts slowly, ensure the runner is ready before workers launch","Check proxy env vars (HTTP_PROXY/HTTPS_PROXY) are not hijacking gRPC traffic; use NO_PROXY for internal hosts"],"exampleFix":"// before\ncc, err := dial(ctx, port.URL, \"data\", 15*time.Second)\nif err != nil { return nil, errors.Wrapf(err, \"failed to connect to data service at %v\", port.URL) }\n// after\nif err := waitForEndpoint(port.URL, 30*time.Second); err != nil {\n  return nil, errors.Wrapf(err, \"data service %v not reachable\", port.URL)\n}\ncc, err := dial(ctx, port.URL, \"data\", 30*time.Second)","handlingStrategy":"retry","validationCode":"conn, err := net.DialTimeout(\"tcp\", hostFromURL(port.URL), 5*time.Second)\nif err != nil {\n  return fmt.Errorf(\"data service %s unreachable before harness start: %w\", port.URL, err)\n}\nconn.Close()","typeGuard":null,"tryCatchPattern":"err := startWorker(ctx)\nif err != nil && isDialFailure(err) {\n  // exponential backoff retry, then fail with clear connectivity message\n  return retryWithBackoff(ctx, startWorker, 3)\n}","preventionTips":["Pre-flight check gRPC endpoint reachability before launching workers","Keep runner and workers on the same network namespace/VPC","Scrub HTTP(S)_PROXY vars or set NO_PROXY for internal hosts","Keep runner and SDK versions aligned"],"tags":["go","apache-beam","grpc","network"],"backgroundTag":"connection-refused","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"}