Fix pipeline hang in Prism runner during Reshuffle elision

shunping · shunping · commit e6ed6b14f755 · 2026-05-20T13:15:59.000-04:00
diff --git a/sdks/go/pkg/beam/runners/prism/internal/handlerunner.go b/sdks/go/pkg/beam/runners/prism/internal/handlerunner.go
@@ -210,7 +210,7 @@ func (h *runner) handleReshuffle(tid string, t *pipepb.PTransform, comps *pipepb
 	}
 
 	// And all the sub transforms.
-	toRemove = append(toRemove, t.GetSubtransforms()...)
+	toRemove = append(toRemove, removeSubTransforms(comps, t.GetSubtransforms())...)
 
 	// Return the new components which is the transforms consumer
 	return prepareResult{

Original file line number	Diff line number	Diff line change
`@@ -210,7 +210,7 @@ func (h runner) handleReshuffle(tid string, t pipepb.PTransform, comps *pipepb`
`210`	`210`	`}`
`211`	`211`
`212`	`212`	`// And all the sub transforms.`
`213`		`- toRemove = append(toRemove, t.GetSubtransforms()...)`
	`213`	`+ toRemove = append(toRemove, removeSubTransforms(comps, t.GetSubtransforms())...)`
`214`	`214`
`215`	`215`	`// Return the new components which is the transforms consumer`
`216`	`216`	`return prepareResult{`