apache/beam · error

job with id %v not found

Error message

job with id %v not found

What it means

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.

Source

Thrown at sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go:439

			State: state,
		}, nil
	}
	job.SendMsg("canceling " + job.String())
	job.Canceling()
	job.CancelFn(ErrCancel)
	return &jobpb.CancelJobResponse{
		State: jobpb.JobState_CANCELLING,
	}, nil
}

// GetMessageStream subscribes to a stream of state changes and messages from the job. If throughput
// is high, this may cause losses of messages.
func (s *Server) GetMessageStream(req *jobpb.JobMessagesRequest, stream jobpb.JobService_GetMessageStreamServer) error {
	s.mu.Lock()
	job, ok := s.jobs[req.GetJobId()]
	s.mu.Unlock()
	if !ok {
		return fmt.Errorf("job with id %v not found", req.GetJobId())
	}

	job.streamCond.L.Lock()
	defer job.streamCond.L.Unlock()
	curMsg := job.minMsg
	curState := job.stateIdx

	state := job.state.Load().(jobpb.JobState_Enum)
	for {
		for (curMsg >= job.maxMsg || len(job.msgs) == 0) && curState > job.stateIdx {
			switch state {
			case jobpb.JobState_CANCELLED, jobpb.JobState_DONE, jobpb.JobState_DRAINED, jobpb.JobState_UPDATED:
				// Reached terminal state.
				return nil
			case jobpb.JobState_FAILED:
				// Ensure we send an error message with the cause of the job failure.
				stream.Send(&jobpb.JobMessagesResponse{
					Response: &jobpb.JobMessagesResponse_MessageResponse{

View on GitHub (pinned to 12126d8942)

Solutions

1. Verify the job ID comes from the current PrepareJob/Run response, not an old run.
2. Re-prepare and run the job if the prism server has restarted.
3. Check server logs for the job's lifecycle to confirm when it was removed.
4. Handle the error client-side by re-establishing the job rather than retrying the same ID.

// Before subscribing, confirm the job exists
if jobID == "" || jobID != currentRun.JobID {
    return errors.New("stale job id; re-run the pipeline")
}

err := client.GetMessageStream(ctx, req)
if err != nil && strings.Contains(err.Error(), "not found") {
    // treat as terminal: re-prepare and re-run the job
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• GetJobMetrics: unknown jobID: %v
• Job {} does not exist
• endpoint not defined
• failed to read metadata from context
• error creating local job server: %v

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