Eliaaazzz opened a new pull request, #39331: URL: https://github.com/apache/beam/pull/39331
Reading a splittable DoFn that self-checkpoints has never worked on the portable Spark runner: no bundle checkpoint handler was registered, so the SDK's first residual raised `UnsupportedOperationException: The ActiveBundle does not have a registered bundle checkpoint handler` and the read died on its first bundle. This is why every SDF test is excluded/skipped for Spark. Fixes #19468. Also addresses #19517 (bundle finalization). ### Batch `SparkExecutableStageFunction` registers a checkpoint handler that collects residual roots, then re-feeds them in fresh bundles until the SDK returns none, resuming each residual at its own requested time. A bounded restriction always runs out, so the loop terminates. An unbounded residual would loop forever in a batch task, so it is rejected with a clear error rather than drained; this keeps the previous fail-fast behavior for that case. The stage also registers `BundleFinalizationHandlers.inMemoryFinalizer`, so `@ProcessElement` bundle finalization callbacks now run (#19517). ### Streaming Residuals leave the stage on a reserved union tag instead of being drained in place, and `SdfResidualRelay` relays them on the driver: each micro-batch's residuals are collected, held until their requested resume time, and fed back as stage input in a later micro-batch, so the read advances across micro-batches instead of occupying one task forever. The relay derives the stage's watermark from the residuals' output watermarks, which is what lets downstream event-time windows fire, and bounds it by the stage's upstream sources so it can never outrun live input. Relays are keyed per job so concurrent jobs in one job server stay isolated. ### Impulse watermark reporting `translateImpulse` reported its watermark once. `GlobalWatermarkHolder` drops sources that stop reporting, and `SparkTimerInternals.forStreamFromSources` requires the sources of a batch to share a synchronized processing time. A single report therefore left the impulse invisible to downstream group operations, and conflicting with any per-batch source flattened alongside it, which fails with `Synchronized time is expected to keep synchronized across sources`. An impulse now reports `+infinity` every micro-batch. This is outside the SDF fix, but the relay cannot coexist with impulse-derived streams in a `GroupByKey` without it, and it is what any `assert_that` on a streaming pipeline hits. ### Testing - Unskips the SDF and bundle finalization tests in `spark_runner_test.py`; 9 pass, 0 fail. - Removes `UsesBoundedSplittableParDo` and `UsesBundleFinalizer` from the batch portable VR excludes. `UsesUnboundedSplittableParDo` stays excluded for now, see below. - New unit tests cover the streaming residual emission and the unbounded-in-batch rejection. - Manually: a Python `UnboundedSource` read streams continuously on a local job server and on a standalone cluster with a separate worker; `FixedWindows(10s)` + `GroupByKey` fires with exact per-window counts; a 34 minute soak held steady throughput with no memory growth. I could not run the Java validates-runner suites locally (Windows), so I am relying on the CI triggers in this PR for that signal. ### Open questions for reviewers - `test_unbounded_source_read` stays skipped, but no longer for SDF reasons: the source self-terminates while a streaming pipeline here runs until its streaming timeout, so the pipeline never reports completion. Worth a separate issue? - Residuals taken for a micro-batch are retired when that batch's collection job reports back, not after every output for the batch has succeeded. Since a failed job stops the streaming context and there is no driver recovery on this path, I judged this not worth the extra machinery, but I would like a second opinion. - Streaming still excludes `UsesUnboundedSplittableParDo`; I would rather enable it in a follow-up once this design has been reviewed. - Happy to split this into a batch-only PR plus a streaming follow-up if that is easier to review. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
