apache/beam · error

unknown environment[ ]

Error message

unknown environment[%v]

What it means

executePipeline iterates pipeline components and handles each as a PTransform, environment (wk.Env), etc. If a component's environment ID does not correspond to any environment handled (it falls into the default case), the runner rejects the pipeline with 'unknown environment[<id>]'. This is a defensive check against pipeline protos referencing environments prism cannot interpret.

Solutions

1. Check the pipeline proto's environments list — the referenced environment ID must exist in components.GetEnvironments().
2. Regenerate the pipeline graph with a current SDK version so environments are emitted correctly.
3. For cross-language jobs, confirm the expansion service registered the environment before submission.

Example fix

// before (pipeline proto missing the env entry)
transform.environment_id = "env-42" // not in environments

// after
// include env-42 in pipeline.components.environments before submitting

// Ensure every transform's environment ID resolves before submitting
envs := comps.GetEnvironments()
for _, t := range comps.GetTransforms() {
    if id := t.GetEnvironmentId(); id != "" {
        if _, ok := envs[id]; !ok {
            log.Fatalf("transform %q references unknown environment %q", t.GetUniqueName(), id)
        }
    }
}

if err := beamx.Run(ctx, p); err != nil && strings.Contains(err.Error(), "unknown environment") {
    log.Fatalf("pipeline references an unregistered environment: %v", err)
}

When it happens

Trigger: Submitting a pipeline whose transform's GetEnvironmentId() references an environment ID absent from, or unregistered in, the pipeline's components map.

Common situations: Hand-crafted or partially deserialized pipeline protos; cross-language pipelines where the environment registry entry was dropped; SDK emitting a new environment kind an older prism doesn't recognize.

Understand the failure class

Background: "Not found" and "does not exist" errors: why "Task not found", "No such folder", and "Can't find" fire when a lookup comes back empty — this error's family across 14 libraries.

Related errors

• couldn't connect to docker
• environment with urn unimplemented
• failed to start environment
• forceLpCoders: coder
• lpUnknownCoders: coder

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

Appendix: source

Thrown at sdks/go/pkg/beam/runners/prism/internal/execute.go:341

				return fmt.Errorf("prism error building stage %v: \n%w", stage.ID, err)
			}
			stages[stage.ID] = stage
			outputs := maps.Keys(stage.OutputsToCoders)
			sort.Strings(outputs)
			em.AddStage(stage.ID, []string{stage.primaryInput}, outputs, stage.sideInputs)
			if stage.stateful {
				em.StageStateful(stage.ID, stage.stateTypeLen)
			}
			if stage.onWindowExpiration.TimerFamily != "" {
				slog.Debug("OnWindowExpiration", slog.String("stage", stage.ID), slog.Any("values", stage.onWindowExpiration))
				em.StageOnWindowExpiration(stage.ID, stage.onWindowExpiration)
			}
			if len(stage.processingTimeTimers) > 0 {
				em.StageProcessingTimeTimers(stage.ID, stage.processingTimeTimers)
			}
			stage.sdfSplittable = config.EnableSDFSplit
		default:
			return fmt.Errorf("unknown environment[%v]", t.GetEnvironmentId())
		}
	}

	// Prime the initial impulses, since we now know what consumes them.
	for _, id := range impulses {
		em.Impulse(id)
	}

	// Use an errgroup to limit max parallelism for the pipeline.
	eg, egctx := errgroup.WithContext(ctx)
	eg.SetLimit(8)

	var instID uint64
	bundles := em.Bundles(egctx, j.CancelFn, func() string {
		return fmt.Sprintf("inst%03d", atomic.AddUint64(&instID, 1))
	})

	// Create a new ticker that fires every 60 seconds.

View on GitHub (pinned to 12126d8942)