apache/beam · error

presultErr.Error()

Error message

presultErr.Error()

What it means

In the Dataflow runner's Execute, after WaitForCompletion the SDK builds a pipeline result via newDataflowPipelineResult. If that call fails (presultErr) while a wait error (err) also exists, both errors are wrapped together and returned alongside a partial result; otherwise presultErr is returned directly. This surfaces failures in fetching final job data/metrics as part of the pipeline execution error.

Solutions

1. Inspect the wrapped cause (presultErr.Error()) to see why the result could not be built — usually a GetMetrics API failure
2. Retry the pipeline result retrieval or re-run; transient Dataflow API failures are common
3. Check the service account has dataflow.jobs.get permission on the project
4. Check whether the job itself failed (err) — fix the root job failure first if present

res, err := beamx.Execute(ctx, p, opts)
if err != nil {
    var presult beam.PipelineResult
    log.Printf("pipeline execution/result error: %v", err)
    // err may wrap both the job error and the result-construction error
    _ = presult
}

When it happens

Trigger: Executing a Dataflow job whose post-completion result construction fails: newDataflowPipelineResult returns an error (e.g. GetMetrics failing, see error 5327), with or without a concurrent WaitForCompletion error.

Common situations: Dataflow API returning errors when fetching job metrics after termination (transient 5xx, permission issues on the job's project), network interruptions between job completion and metrics fetch, or the job being deleted/soft-deleted before metrics can be read.

Related errors

• failed to get job
• failed to get metrics
• bad I
• bad KV
• bad N

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

Appendix: source

Thrown at sdks/go/pkg/beam/runners/dataflow/dataflowlib/execute.go:149

	if endpoint == "" {
		log.Infof(ctx, "Console: https://console.cloud.google.com/dataflow/jobs/%v/%v?project=%v", opts.Region, upd.Id, opts.Project)
	}
	log.Infof(ctx, "Logs: https://console.cloud.google.com/logs/viewer?project=%v&resource=dataflow_step%%2Fjob_id%%2F%v", opts.Project, upd.Id)

	presult.jobID = upd.Id

	if async {
		return presult, nil
	}

	// (4) Wait for completion.
	err = WaitForCompletion(ctx, client, opts.Project, opts.Region, upd.Id)

	res, presultErr := newDataflowPipelineResult(ctx, client, raw, opts.Project, opts.Region, upd.Id)
	if presultErr != nil {
		if err != nil {
			return presult, errors.Wrap(err, presultErr.Error())
		}
		return presult, presultErr
	}
	return res, err
}

// PrintJob logs the Dataflow job.
func PrintJob(ctx context.Context, job *df.Job) {
	str, err := json.MarshalIndent(job, "", "  ")
	if err != nil {
		log.Infof(ctx, "Failed to print job %v: %v", job.Id, err)
	}
	log.Info(ctx, string(str))
}

type dataflowPipelineResult struct {
	jobID   string
	metrics *metrics.Results

View on GitHub (pinned to 12126d8942)