{"record":{"id":"6f14e97143cfec8d","repo":"apache/beam","slug":"failed-to-get-job-stream","errorCode":null,"errorMessage":"failed to get job stream","messagePattern":"failed to get job stream","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/universal/runnerlib/job.go","lineNumber":103,"sourceCode":"func Submit(ctx context.Context, client jobpb.JobServiceClient, id, token string) (string, error) {\n\treq := &jobpb.RunJobRequest{\n\t\tPreparationId:  id,\n\t\tRetrievalToken: token,\n\t}\n\n\tresp, err := client.Run(ctx, req)\n\tif err != nil {\n\t\treturn \"\", errors.Wrap(err, \"failed to submit job\")\n\t}\n\treturn resp.GetJobId(), nil\n}\n\n// WaitForCompletion monitors the given job until completion. It logs any messages\n// and state changes received.\nfunc WaitForCompletion(ctx context.Context, client jobpb.JobServiceClient, jobID string) error {\n\tstream, err := client.GetMessageStream(ctx, &jobpb.JobMessagesRequest{JobId: jobID})\n\tif err != nil {\n\t\treturn errors.Wrap(err, \"failed to get job stream\")\n\t}\n\n\tmostRecentError := \"<no error received>\"\n\tvar errReceived, jobFailed bool\n\n\tfor {\n\t\tmsg, err := stream.Recv()\n\t\tif err != nil {\n\t\t\tif err == io.EOF {\n\t\t\t\tif jobFailed {\n\t\t\t\t\t// Connection finished, so time to exit, produce what we have.\n\t\t\t\t\treturn errors.Errorf(\"job %v failed:\\n%v\", jobID, mostRecentError)\n\t\t\t\t}\n\t\t\t\treturn nil\n\t\t\t}\n\t\t\treturn err\n\t\t}\n","sourceCodeStart":85,"sourceCodeEnd":121,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/universal/runnerlib/job.go#L85-L121","documentation":"WaitForCompletion monitors a running job by opening the runner's GetMessageStream gRPC stream. If opening the stream fails, the error is wrapped as \"failed to get job stream\". Without the message stream, the client cannot observe state changes or logs for the job.","triggerScenarios":"client.GetMessageStream(ctx, &jobpb.JobMessagesRequest{JobId: jobID}) returns an error immediately after job submission, called from Execute.","commonSituations":"Network interruption right after submission; runner rejects the job id; gRPC/TLS misconfiguration; runner restarts between Submit and WaitForCompletion.","solutions":["Check the wrapped gRPC error for connectivity vs. job-id issues.","Verify the runner endpoint and TLS settings.","Check the job's state on the runner (it may still be running without a message stream).","Retry the job monitoring stream or resubmit the job."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"if err := beam.Run(ctx, runner, p); err != nil && strings.Contains(err.Error(), \"failed to get job stream\") {\n    // check job state via GetJobState RPC or the runner UI before resubmitting\n}","preventionTips":["Ensure a stable network connection to the runner during job execution.","Avoid aggressive context deadlines that can kill long-lived message streams.","Verify TLS/proxy settings that may break gRPC streaming.","Check job existence on the runner if the id may be stale."],"tags":["go","beam","grpc","streaming"],"backgroundTag":"network-request-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"}