{"record":{"id":"05b811967eaa49bc","repo":"apache/beam","slug":"sdk-side-reshuffle-not-yet-supported","errorCode":null,"errorMessage":"SDK side reshuffle not yet supported","messagePattern":"SDK side reshuffle not yet supported","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/handlerunner.go","lineNumber":169,"sourceCode":"\t\t// Return the new components which is the transforms consumer\n\t\treturn prepareResult{\n\t\t\t// We sub this flatten with itself, to not drop it.\n\t\t\tSubbedComps: &pipepb.Components{\n\t\t\t\tTransforms:   tSubs,\n\t\t\t\tPcollections: pcollSubs,\n\t\t\t},\n\t\t\tRemovedLeaves: nil,\n\t\t\tForcedRoots:   forcedRoots,\n\t\t}\n\t}\n\treturn prepareResult{}\n}\n\nfunc (h *runner) handleReshuffle(tid string, t *pipepb.PTransform, comps *pipepb.Components) prepareResult {\n\t// TODO: Implement the windowing strategy the \"backup\" transforms used for Reshuffle.\n\n\tif h.config.SDKReshuffle {\n\t\tpanic(\"SDK side reshuffle not yet supported\")\n\t}\n\n\t// A Reshuffle, in principle, is a no-op on the pipeline structure, WRT correctness.\n\t// It could however affect performance, so it exists to tell the runner that this\n\t// point in the pipeline needs a fusion break, to enable the pipeline to change it's\n\t// degree of parallelism.\n\t//\n\t// The change of parallelism goes both ways. It could allow for larger batch sizes\n\t// enable smaller batch sizes downstream if it is infact paralleizable.\n\t//\n\t// But for a single transform node per stage runner, we can elide it entirely,\n\t// since the input collection and output collection types match.\n\n\t// Get the input and output PCollections, there should only be 1 each.\n\tif len(t.GetInputs()) != 1 {\n\t\tpanic(\"Expected single input PCollection in reshuffle: \" + prototext.Format(t))\n\t}\n\tif len(t.GetOutputs()) != 1 {","sourceCodeStart":151,"sourceCodeEnd":187,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/handlerunner.go#L151-L187","documentation":"handleReshuffle rejects reshuffles that the SDK has already expanded into backup transforms (h.config.SDKReshuffle true). Prism only implements its own runner-side reshuffle (a fusion break); it panics when asked to handle an SDK-side reshuffle expansion instead.","triggerScenarios":"A pipeline is submitted with config.SDKReshuffle enabled while the graph contains a Reshuffle/Redistribute transform that was expanded on the SDK side into its backup group-by-key transforms.","commonSituations":"Enabling the SDKReshuffle prism config flag (usually intended to test SDK-side reshuffle behavior) in a normal pipeline run; combining an SDK build that expands reshuffle with prism's unsupported handling path.","solutions":["Remove the SDKReshuffle setting (or set it false) so prism handles Reshuffle natively as a fusion break.","Rebuild the pipeline without SDK-side reshuffle expansion so the Reshuffle URN reaches the runner intact.","If SDK-side reshuffle is required, use a runner that supports it (e.g. Flink/Dataflow) or implement runner-side support in prism."],"exampleFix":"// before\nrunnerOpts := []func(*prism.Option){ prism.SDKReshuffle(true) }\n// after\nrunnerOpts := []func(*prism.Option){ prism.SDKReshuffle(false) }","handlingStrategy":"validation","validationCode":"// ensure prism options don't enable SDKReshuffle\nif sdkReshuffleEnabled { /* route to a runner that supports SDK-side reshuffle */ }","typeGuard":null,"tryCatchPattern":"defer func(){ if r := recover(); r != nil && strings.Contains(fmt.Sprint(r), \"SDK side reshuffle not yet supported\") { /* switch runner or config */ } }()","preventionTips":["Don't set SDKReshuffle(true) with prism","Use standard beam.Reshuffle() and let prism fuse-break natively"],"tags":["go","apache-beam","prism-runner","reshuffle"],"backgroundTag":"unsupported-operation","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"}