apache/beam · error

bad payload for env

Error message

bad payload for env %v: %v

What it means

When translating a Beam pipeline into a Dataflow job, containerImages iterates the pipeline's environments and expects each environment payload to be a protobuf-encoded DockerPayload. If proto.Unmarshal of an environment's payload fails, translation aborts with 'bad payload for env'. This indicates a malformed or unexpected environment payload in the pipeline graph — typically an SDK/internal inconsistency or corruption.

Solutions

1. Check that all cross-language transforms (Java/Python expansions) use a Beam SDK version compatible with your Go SDK — upgrade both sides to matching versions.
2. Inspect which environment id fails (it's in the message) and trace which transform registered it; replace or update that transform.
3. If you construct pipeline environments manually, ensure the payload is proto.Marshal(&pipepb.DockerPayload{...}) of the correct type.
4. Re-run the expansion service and confirm its logs show a valid docker payload; regenerate any custom expansion artifacts.
5. Report/inspect if caused by an SDK bug: dump the pipeline proto before translation to confirm payload contents.

Example fix

// before: env registered with wrong payload type
env.Payload = proto.Marshal(&pipepb.ProcessPayload{...})
// bad payload for env env1: proto: cannot parse invalid wire-format data

// after
env.Payload, _ = proto.Marshal(&pipepb.DockerPayload{
    DockerImage: "gcr.io/project/beam_go_sdk:2.x.x",
    Capabilities: caps,
})

var probe pipepb.DockerPayload
for id, env := range p.GetComponents().GetEnvironments() {
    if err := proto.Unmarshal(env.GetPayload(), &probe); err != nil {
        log.Printf("environment %s has non-Docker payload before submit", id)
    }
}

func isDockerEnv(env *pipepb.Environment) bool {
    var dp pipepb.DockerPayload
    return proto.Unmarshal(env.GetPayload(), &dp) == nil
}

// Wrap job submission so translation errors surface with pipeline details:
if _, _, err := dataflowlib.Translate(ctx, p, opts); err != nil {
    if strings.Contains(err.Error(), "bad payload for env") {
        log.Fatalf("environment payload mismatch (check xlang SDK versions): %v", err)
    }
    return err
}

When it happens

Trigger: Running a pipeline via the Dataflow runner (dataflowlib.Translate) where an environment's payload bytes cannot be unmarshalled into pipepb.DockerPayload — e.g. an environment created with a non-Docker payload type by a custom expansion or a cross-language transform producing an incompatible payload.

Common situations: Cross-language (xlang) expansions registering environments with external payload types; mismatched beam versions between the Go SDK and expansion service; custom pipeline injection that populates environments manually.

Understand the failure class

Background: "cannot parse invalid wire-format data", "cannot unmarshal", "failed unmarshalling": protobuf unmarshal errors explained — this error's family across 10 libraries.

Related errors

• invalid transform payload
• Unsupported access pattern for %r: %r
• A non-standard version of Beam SDK detected
• Aliased enumerations not currently supported.
• Any not yet supported

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

Appendix: source

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

	// Worker is the worker binary override.
	Worker string

	// WorkerHash is the SHA-256 hash of the worker binary.
	WorkerHash string

	// -- Internal use only. Not supported in public Dataflow. --

	TeardownPolicy string
}

func containerImages(p *pipepb.Pipeline) ([]*df.SdkHarnessContainerImage, []string, error) {
	envs := p.GetComponents().GetEnvironments()
	ret := make([]*df.SdkHarnessContainerImage, 0, len(envs))
	display := make([]string, 0, len(envs))
	for id, env := range envs {
		var payload pipepb.DockerPayload
		if err := proto.Unmarshal(env.GetPayload(), &payload); err != nil {
			return nil, nil, fmt.Errorf("bad payload for env %v: %v", id, err)
		}
		singleCore := true
		for _, c := range env.GetCapabilities() {
			if c == graphx.URNMultiCore {
				singleCore = false
			}
		}
		ret = append(ret, &df.SdkHarnessContainerImage{
			ContainerImage:            payload.GetContainerImage(),
			UseSingleCorePerContainer: singleCore,
			Capabilities:              env.GetCapabilities(),
			EnvironmentId:             id,
		})
		display = append(display, payload.GetContainerImage())
	}
	return ret, display, nil
}

View on GitHub (pinned to 12126d8942)