{"record":{"id":"5d64f9986474e3d7","repo":"apache/beam","slug":"bad-payload-for-env-v-v","errorCode":null,"errorMessage":"bad payload for env %v: %v","messagePattern":"bad payload for env (.+?): (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/dataflow/dataflowlib/job.go","lineNumber":103,"sourceCode":"\t// Worker is the worker binary override.\n\tWorker string\n\n\t// WorkerHash is the SHA-256 hash of the worker binary.\n\tWorkerHash string\n\n\t// -- Internal use only. Not supported in public Dataflow. --\n\n\tTeardownPolicy string\n}\n\nfunc containerImages(p *pipepb.Pipeline) ([]*df.SdkHarnessContainerImage, []string, error) {\n\tenvs := p.GetComponents().GetEnvironments()\n\tret := make([]*df.SdkHarnessContainerImage, 0, len(envs))\n\tdisplay := make([]string, 0, len(envs))\n\tfor id, env := range envs {\n\t\tvar payload pipepb.DockerPayload\n\t\tif err := proto.Unmarshal(env.GetPayload(), &payload); err != nil {\n\t\t\treturn nil, nil, fmt.Errorf(\"bad payload for env %v: %v\", id, err)\n\t\t}\n\t\tsingleCore := true\n\t\tfor _, c := range env.GetCapabilities() {\n\t\t\tif c == graphx.URNMultiCore {\n\t\t\t\tsingleCore = false\n\t\t\t}\n\t\t}\n\t\tret = append(ret, &df.SdkHarnessContainerImage{\n\t\t\tContainerImage:            payload.GetContainerImage(),\n\t\t\tUseSingleCorePerContainer: singleCore,\n\t\t\tCapabilities:              env.GetCapabilities(),\n\t\t\tEnvironmentId:             id,\n\t\t})\n\t\tdisplay = append(display, payload.GetContainerImage())\n\t}\n\treturn ret, display, nil\n}\n","sourceCodeStart":85,"sourceCodeEnd":121,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/dataflow/dataflowlib/job.go#L85-L121","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["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.","Inspect which environment id fails (it's in the message) and trace which transform registered it; replace or update that transform.","If you construct pipeline environments manually, ensure the payload is proto.Marshal(&pipepb.DockerPayload{...}) of the correct type.","Re-run the expansion service and confirm its logs show a valid docker payload; regenerate any custom expansion artifacts.","Report/inspect if caused by an SDK bug: dump the pipeline proto before translation to confirm payload contents."],"exampleFix":"// before: env registered with wrong payload type\nenv.Payload = proto.Marshal(&pipepb.ProcessPayload{...})\n// bad payload for env env1: proto: cannot parse invalid wire-format data\n\n// after\nenv.Payload, _ = proto.Marshal(&pipepb.DockerPayload{\n    DockerImage: \"gcr.io/project/beam_go_sdk:2.x.x\",\n    Capabilities: caps,\n})","handlingStrategy":"try-catch","validationCode":"var probe pipepb.DockerPayload\nfor id, env := range p.GetComponents().GetEnvironments() {\n    if err := proto.Unmarshal(env.GetPayload(), &probe); err != nil {\n        log.Printf(\"environment %s has non-Docker payload before submit\", id)\n    }\n}","typeGuard":"func isDockerEnv(env *pipepb.Environment) bool {\n    var dp pipepb.DockerPayload\n    return proto.Unmarshal(env.GetPayload(), &dp) == nil\n}","tryCatchPattern":"// Wrap job submission so translation errors surface with pipeline details:\nif _, _, err := dataflowlib.Translate(ctx, p, opts); err != nil {\n    if strings.Contains(err.Error(), \"bad payload for env\") {\n        log.Fatalf(\"environment payload mismatch (check xlang SDK versions): %v\", err)\n    }\n    return err\n}","preventionTips":["Keep Go SDK, expansion services, and other language SDKs on the same Beam version.","Don't hand-craft pipeline environments; let the SDK create them.","Log environment payloads before Translate when debugging cross-language pipelines."],"tags":["dataflow","protobuf","pipeline-translation","xlang"],"backgroundTag":"protobuf-unmarshal-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}