{"record":{"id":"ea358b0124f1b96a","repo":"apache/beam","slug":"job-with-id-v-not-found","errorCode":null,"errorMessage":"job with id %v not found","messagePattern":"job with id (.+?) not found","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go","lineNumber":439,"sourceCode":"\t\t\tState: state,\n\t\t}, nil\n\t}\n\tjob.SendMsg(\"canceling \" + job.String())\n\tjob.Canceling()\n\tjob.CancelFn(ErrCancel)\n\treturn &jobpb.CancelJobResponse{\n\t\tState: jobpb.JobState_CANCELLING,\n\t}, nil\n}\n\n// GetMessageStream subscribes to a stream of state changes and messages from the job. If throughput\n// is high, this may cause losses of messages.\nfunc (s *Server) GetMessageStream(req *jobpb.JobMessagesRequest, stream jobpb.JobService_GetMessageStreamServer) error {\n\ts.mu.Lock()\n\tjob, ok := s.jobs[req.GetJobId()]\n\ts.mu.Unlock()\n\tif !ok {\n\t\treturn fmt.Errorf(\"job with id %v not found\", req.GetJobId())\n\t}\n\n\tjob.streamCond.L.Lock()\n\tdefer job.streamCond.L.Unlock()\n\tcurMsg := job.minMsg\n\tcurState := job.stateIdx\n\n\tstate := job.state.Load().(jobpb.JobState_Enum)\n\tfor {\n\t\tfor (curMsg >= job.maxMsg || len(job.msgs) == 0) && curState > job.stateIdx {\n\t\t\tswitch state {\n\t\t\tcase jobpb.JobState_CANCELLED, jobpb.JobState_DONE, jobpb.JobState_DRAINED, jobpb.JobState_UPDATED:\n\t\t\t\t// Reached terminal state.\n\t\t\t\treturn nil\n\t\t\tcase jobpb.JobState_FAILED:\n\t\t\t\t// Ensure we send an error message with the cause of the job failure.\n\t\t\t\tstream.Send(&jobpb.JobMessagesResponse{\n\t\t\t\t\tResponse: &jobpb.JobMessagesResponse_MessageResponse{","sourceCodeStart":421,"sourceCodeEnd":457,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go#L421-L457","documentation":"GetMessageStream looks up the requested job ID in the Server's job map; if absent it returns this error instead of streaming messages. It indicates the client referenced a job that this prism server never prepared or has fully forgotten.","triggerScenarios":"Calling the JobService GetMessageStream RPC with a JobMessagesRequest whose JobId does not exist in s.jobs — typically after server restart or a mistyped/stale job ID.","commonSituations":"A test client keeps polling messages after the server was recreated; job ID captured from a previous run's logs; job was garbage-collected on the server.","solutions":["Verify the job ID comes from the current PrepareJob/Run response, not an old run.","Re-prepare and run the job if the prism server has restarted.","Check server logs for the job's lifecycle to confirm when it was removed.","Handle the error client-side by re-establishing the job rather than retrying the same ID."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// Before subscribing, confirm the job exists\nif jobID == \"\" || jobID != currentRun.JobID {\n    return errors.New(\"stale job id; re-run the pipeline\")\n}","typeGuard":null,"tryCatchPattern":"err := client.GetMessageStream(ctx, req)\nif err != nil && strings.Contains(err.Error(), \"not found\") {\n    // treat as terminal: re-prepare and re-run the job\n}","preventionTips":["Always use the job ID returned by the current Prepare/Run call.","Re-create jobs after any prism server restart.","Log job IDs with their run/session to avoid cross-run mixing."],"tags":["go","beam-prism","grpc","job-management"],"backgroundTag":"resource-not-found","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}