apache/beam · warning

error may be due to Apache Beam Go's migration from the…

Error message

%w
error may be due to Apache Beam Go's migration from the direct runner to the prism runner. While the failure(s) should be fixed, you can continue to use the direct runner with this TestMain override: `func TestMain(m *testing.M) { ptest.MainWithDefault(m, "direct") }`

What it means

ptest.Run wraps beam.Run for pipeline tests and augments errors when the default runner (prism, after the direct runner deprecation) fails while the user never explicitly chose the runner. The wrapped message tells the developer the failure may stem from the direct-to-prism migration and shows a TestMain override to fall back to the direct runner.

Solutions

1. Fix the underlying pipeline failure reported inside the wrapped error
2. Temporarily pin the direct runner via `func TestMain(m *testing.M) { ptest.MainWithDefault(m, "direct") }` in the test package
3. Run the test with an explicit -runner flag to bypass the default (e.g. -runner direct)
4. Compare pipeline behavior between direct and prism runners to identify the incompatibility

Example fix

// before (test file with default prism runner)
func TestMain(m *testing.M) { ptest.Main(m) }
// after (temporary fallback to direct runner)
func TestMain(m *testing.M) { ptest.MainWithDefault(m, "direct") }

if err := ptest.Run(p); err != nil {
    if strings.Contains(err.Error(), "prism runner") {
        // migration hint present: retry with -runner direct or fix pipeline
        t.Fatalf("pipeline failed (possibly prism migration): %v", err)
    }
    t.Fatalf("pipeline failed: %v", err)
}

When it happens

Trigger: Running a ptest-based test (ptest.Run) where the default runner is "prism", the runner was not manually set via the --runner flag, and beam.Run returns an error. The error is then wrapped with the migration hint.

Common situations: Upgrading Beam versions around the v2.5x direct-to-prism runner migration; tests that passed under the direct runner now fail under prism due to behavioral differences (ordering, watermarking, transport).

Understand the failure class

Background: "is deprecated and will be removed" — deprecation warnings for old API names, keywords, and options, and how to migrate before the removal release — this error's family across 29 libraries.

Related errors

• missing entries (missing in actual, present in expected)
• observed PCollection has incompatible type
• passert.Count( ) = , want
• passert.Diff input PColections don't have matching types
• passert.Hash( ) = ( , ), want ( , )

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

Appendix: source

Thrown at sdks/go/pkg/beam/testing/ptest/ptest.go:106

	return defaultRunnerOverride
}

// MainCalled returns true iff Main or MainRet has been called.
func MainCalled() bool {
	return mainCalled
}

// Run runs a pipeline for testing. The semantics of the pipeline is expected
// to be verified through passert.
func Run(p *beam.Pipeline) error {
	r := getRunner()
	_, err := beam.Run(context.Background(), r, p)
	// Until a few versions from now (say, v2.56),
	// if there's an error, and
	// the runner is prism, and it was the set default runner, but not a manually specificed runner via the flag
	// augment the error with instructions to switch back to the direct runner.
	if err != nil && r == "prism" && r == defaultRunnerOverride && r != *Runner {
		err = fmt.Errorf("%w\nerror may be due to Apache Beam Go's migration from the direct runner to the prism runner."+
			" While the failure(s) should be fixed, you can continue to use the direct runner with this TestMain override:"+
			" `func TestMain(m *testing.M) { ptest.MainWithDefault(m, \"direct\") }`", err)
	}
	return err
}

// RunWithMetrics runs a pipeline for testing with that returns metrics.Results
// in the form of Pipeline Result
func RunWithMetrics(p *beam.Pipeline) (beam.PipelineResult, error) {
	return beam.Run(context.Background(), getRunner(), p)
}

// RunAndValidate runs a pipeline for testing and validates the result, failing
// the test if the pipeline fails.
func RunAndValidate(t *testing.T, p *beam.Pipeline) beam.PipelineResult {
	t.Helper()
	pr, err := RunWithMetrics(p)
	if err != nil {

View on GitHub (pinned to 12126d8942)