apache/beam · critical

Job failed

Error message

Job %s failed

What it means

When the Dataflow service reports the job state JOB_STATE_FAILED, currentStateMessage (called from WaitForCompletion) returns a terminal error 'Job <jobID> failed'. This is not an SDK malfunction: the job itself ran and failed on the service, and the SDK propagates that as the pipeline's error. Diagnose via the Dataflow console/logs, not the Go stack.

Solutions

1. Open the job in the Dataflow console (or `gcloud dataflow jobs describe <jobID>`) and read the failure logs / stack traces
2. Fix the root cause in the pipeline: worker errors, invalid input/output paths, or quota limits
3. Check worker logs in Cloud Logging for the failing step and exception
4. Verify quotas (CPU, in-use IPs, disk) and the worker service account's permissions

if err := run(); err != nil && strings.Contains(err.Error(), "failed") {
    var jobID string
    if n, _ := fmt.Sscanf(err.Error(), "Job %s failed", &jobID); n == 1 {
        log.Printf("inspect job at console: %s", jobID)
    }
}

When it happens

Trigger: Executing a pipeline whose Dataflow job transitions to JOB_STATE_FAILED — detected during WaitForCompletion polling and returned as the final error of Execute.

Common situations: Worker crashes (OOM, quota, image pull failures), user code panics inside DoFns, invalid sources/sinks (bad GCS paths, missing BigQuery datasets), networking/VPC issues preventing workers from reaching services, or insufficient service account permissions.

Related errors

• bad I
• bad KV
• bad N
• bad pair: %+v
• Beam has not been initialized. Call beam.Init() before…

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

Appendix: source

Thrown at sdks/go/pkg/beam/runners/dataflow/dataflowlib/job.go:305

// Errors are always terminal.
func currentStateMessage(currentState, jobID string) (bool, string, error) {
	switch currentState {
	// Add all Terminal Success stats here.
	case "JOB_STATE_DONE", "JOB_STATE_CANCELLED", "JOB_STATE_DRAINED", "JOB_STATE_UPDATED":
		var state string
		switch currentState {
		case "JOB_STATE_DONE":
			state = "succeeded!"
		case "JOB_STATE_CANCELLED":
			state = "cancelled"
		case "JOB_STATE_DRAINED":
			state = "drained"
		case "JOB_STATE_UPDATED":
			state = "updated"
		}
		return true, fmt.Sprintf("Job %v %v", jobID, state), nil
	case "JOB_STATE_FAILED":
		return true, "", errors.Errorf("Job %s failed", jobID)
	case "JOB_STATE_RUNNING":
		return false, "Job still running ...", nil
	default:
		return false, fmt.Sprintf("Job state: %v ...", currentState), nil
	}
}

// NewClient creates a new dataflow client with default application credentials
// and CloudPlatformScope. The Dataflow endpoint is optionally overridden.
func NewClient(ctx context.Context, endpoint string) (*df.Service, error) {
	cl, err := google.DefaultClient(ctx, df.CloudPlatformScope)
	if err != nil {
		return nil, err
	}
	client, err := df.New(cl)
	if err != nil {
		return nil, err
	}

View on GitHub (pinned to 12126d8942)