[Prism] Fix SDF splits and checkpoints in ElementManager.ReturnResiduals

Fixes a bug where splittable DoFN (SDF) splits and checkpoints would completely lose
their deferred residual restrictions, causing the pipeline to terminate prematurely
and downstream assertions to fail due to missing elements (e.g., test_register_finalizations).

Root Cause:
- The previous patch in v2 fixed a double-counting bug of livePending by removing the
  decoding and rescheduling of split residuals (stage.AddPending and em.addPending)
  from ReturnResiduals(), assuming they were already placed back by stage.splitBundle().
- This holds true for normal non-SDF channel splits, where stage.splitBundle already
  puts the unprocessed original elements back.
- However, when a splittable DoFn (SDF) checkpoints itself, the active element splits on
  its restriction rather than simple unprocessed channel elements. In this case, the original
  remaining elements (res) in splitBundle() has a length of 0, but the SDK worker
  returns a new restriction in the residuals.Data (e.g. unprocessedElements length 1).
- Because the previous patch completely removed rescheduling from ReturnResiduals, this new
  residual restriction was completely lost and never added back to the pending queue.

Solution:
- Inside ReturnResiduals(), we dynamically calculate the original remaining elements in
  the bundle: originalRemainingCount := len(completed.es) - firstRsIndex.
- We compare the total returned residuals (unprocessedElements) against originalRemainingCount.
- If len(unprocessedElements) > originalRemainingCount, the difference represents the new SDF
  residual restrictions. We selectively add ONLY these new residuals back to the stage pending
  heap and safely increment em.livePending by this difference.
- This elegantly preserves the fix for normal channel splits (preventing double-counting), while
  ensuring SDF checkpoint residuals are correctly scheduled.
- Also includes detailed slog.Info logging during execution to track livePending state changes
  accurately in addPending, ReturnResiduals, and splitBundle.

Author: shunping

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

@@ -240,8 +240,10 @@ type ElementManager struct {
 }
 
 func (em *ElementManager) addPending(v int) {
+	prev := em.livePending.Load()
 	em.livePending.Add(int64(v))
 	em.pendingElements.Add(v)
+	slog.Info("em.addPending", "delta", v, "prev", prev, "current", em.livePending.Load())
 }
 
 // LinkID represents a fully qualified input or output.
@@ -1091,16 +1093,25 @@ func (em *ElementManager) FailBundle(rb RunBundle) {
 	em.markChangedAndClearBundle(rb.StageID, rb.BundleID, nil)
 }
 
-// ReturnResiduals is called after a successful split, so the remaining work
-// can be re-assigned to a new bundle.
 func (em *ElementManager) ReturnResiduals(rb RunBundle, firstRsIndex int, inputInfo PColInfo, residuals Residuals) {
 	stage := em.stages[rb.StageID]
 
+	slog.Info("ElementManager.ReturnResiduals start", "bundle", rb, "firstRsIndex", firstRsIndex)
+	
+	stage.mu.Lock()
+	completed := stage.inprogress[rb.BundleID]
+	originalRemainingCount := len(completed.es) - firstRsIndex
+	stage.mu.Unlock()
+
 	stage.splitBundle(rb, firstRsIndex, em)
 	unprocessedElements := reElementResiduals(residuals.Data, inputInfo, rb)
-	if len(unprocessedElements) > 0 {
-		slog.Debug("ReturnResiduals: unprocessed elements", "bundle", rb, "count", len(unprocessedElements))
+	if len(unprocessedElements) > originalRemainingCount {
+		newResiduals := unprocessedElements[originalRemainingCount:]
+		slog.Info("ReturnResiduals: new residuals added back", "bundle", rb, "count", len(newResiduals))
+		count := stage.AddPending(em, newResiduals)
+		em.addPending(count)
 	}
+	slog.Info("ElementManager.ReturnResiduals end", "bundle", rb, "unprocessedCount", len(unprocessedElements), "livePending", em.livePending.Load())
 	em.markStagesAsChanged(singleSet(rb.StageID))
 }
 
@@ -2185,7 +2196,7 @@ func (ss *stageState) splitBundle(rb RunBundle, firstResidual int, em *ElementMa
 	defer ss.mu.Unlock()
 
 	es := ss.inprogress[rb.BundleID]
-	slog.Debug("split elements", "bundle", rb, "elem count", len(es.es), "res", firstResidual)
+	slog.Info("splitBundle start", "bundle", rb, "elem count", len(es.es), "firstResidual", firstResidual, "livePending", em.livePending.Load())
 
 	prim := es.es[:firstResidual]
 	res := es.es[firstResidual:]
@@ -2205,6 +2216,7 @@ func (ss *stageState) splitBundle(rb RunBundle, firstResidual int, em *ElementMa
 	// we don't need to increment pending count in em, since it is already pending
 	ss.kind.addPending(ss, em, res)
 	ss.inprogress[rb.BundleID] = es
+	slog.Info("splitBundle completed", "bundle", rb, "primaryCount", len(prim), "residualCount", len(res), "livePending", em.livePending.Load())
 }
 
 // minimumPendingTimestamp returns the minimum pending timestamp from all pending elements,