[damccorm]

Note: this PR was auto-generated by tooling. Before moving it out of draft, the author has done a full review of the PR, specifically evaluating it on the following criteria:

• The PR is an appropriate fix for the issue (e.g. we're not changing a test to avoid a bug)
• The PR is a simple fix to the issue (e.g. we're not adding infra when we could just modify a test)
• The PR is not covering for other problems (e.g. if we're bumping timeouts, it is not because we're covering for a performance problem)

Full details of the methodology of generation and review can be found here - https://docs.google.com/document/d/14lj7ECVov4T4EdwmIzNlWrRGE3lLDMOwX5xUMGzaLoc/edit?usp=sharing

If you have concerns with the PR or with the underlying methodology used to generate this, please comment below - they will be addressed by the author not the tooling :)

Summary

Fixes a watermark synchronization race in Spark Runner DStream streaming pipelines (PostCommit Java PVR Spark4 Streaming, e.g. https://github.com/apache/beam/actions/runs/43374186705/job/131716702201) where CombineTest$BasicTests.testBinaryCombinePerKeyInstanceMethodReference failed with:

java.lang.AssertionError: Combine.perKey(Anonymous)/Combine.GroupedValues/ParDo(Anonymous)/ParMultiDo(Anonymous).output:
Expected: iterable with items [<KV{a, 4}>, <KV{b, 2}>, <KV{c, 4}>] in any order
     but: no item matches: <KV{a, 4}>, <KV{b, 2}>, <KV{c, 4}> in []

Production Impact

Impacts real Spark Runner streaming pipelines (both Portable DStream streaming and Classic streaming pipelines using Flatten / Impulse via SingleEmitInputDStream).

• Who is affected: Users running streaming pipelines on the Spark Runner (DStream backend) with multi-stage GroupByKey / Combine.perKey topologies where a micro-batch takes longer than batchIntervalMillis (default 1000ms), or running multiple pipelines sequentially in the same JVM via SparkPipelineRunner.
• What happens:
  1. Unlike TestDStream (waitForLastBatch) and WatermarkSyncedDStream (awaitWatermarkSyncWith), SingleEmitInputDStream.compute(Time) did not wait for the previous micro-batch's watermark advancement (GlobalWatermarkHolder.getLastWatermarkedBatchTime()) before generating the next micro-batch's RDD. When Batch 1 takes longer than batchIntervalMillis (e.g., while initializing the SDK harness), Spark Streaming's JobGenerator queues Batch 2 while Batch 1 is still executing. As soon as Batch 1's Spark job finishes, JobScheduler.jobExecutor starts executing Batch 2 concurrently with WatermarkAdvancingStreamingListener.onBatchCompleted(Batch 1) running on Spark's asynchronous LiveListenerBus thread.
  2. Because SparkGroupAlsoByWindowViaWindowSet fetches GlobalWatermarkHolder.get(...) lazily inside each stage on the executor, an upstream GroupByKey stage (Combine.perKey) in Batch 2 can read GlobalWatermarkHolder.get(...) before advance() completes (seeing TIMESTAMP_MIN_VALUE and not firing its buffered elements), while a downstream GroupByKey stage (PAssert) in the same Batch 2 reads GlobalWatermarkHolder.get(...) milliseconds later after advance() completes (seeing TIMESTAMP_MAX_VALUE and firing premature timers before upstream results arrive).
  3. Additionally, SparkPipelineRunner.run did not call GlobalWatermarkHolder.clear() or MetricsAccumulator.clear() (unlike TestSparkRunner.run), leaving static watermark and metrics state dirty across sequential pipeline runs in the same JVM, and GlobalWatermarkHolder.add / get were not synchronized with advance.
• How this fix resolves it:
  1. SingleEmitInputDStream.compute(Time) now waits for GlobalWatermarkHolder.getLastWatermarkedBatchTime() >= lastValidTimeMs when watermarks are registered (matching TestDStream), ensuring Batch N+1 is not generated until Batch N's watermark update has completed.
  2. Synchronizes GlobalWatermarkHolder.add, addAll, get, and hasWatermarks on GlobalWatermarkHolder.class, and resets watermarkCache in clear().
  3. Clears MetricsAccumulator and GlobalWatermarkHolder in SparkPipelineRunner.run before starting a pipeline.

Verification

• Triggered beam_PostCommit_Java_PVR_Spark4_Streaming.yml workflow run on branch: 43388934435

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR resolves a race condition in portable Spark Streaming where subsequent micro-batches could execute before WatermarkAdvancingStreamingListener finished advancing watermarks for the prior batch, causing downstream stages in multi-GBK topologies to observe inconsistent watermarks. It also properly resets GlobalWatermarkHolder and MetricsAccumulator state in SparkPipelineRunner between sequential runs and synchronizes watermark state access in GlobalWatermarkHolder.

🔍 General Feedback

• Synchronizing SingleEmitInputDStream.compute(Time) against GlobalWatermarkHolder.getLastWatermarkedBatchTime() (matching TestDStream's pattern) and guarding on GlobalWatermarkHolder.hasWatermarks() cleanly prevents out-of-order watermark reads across stages without hanging non-watermarked streams or checkpoint recovery tests.
• Synchronizing add, addAll, hasWatermarks, and get in GlobalWatermarkHolder and resetting watermarkCache in clear() closes thread-safety and state-leakage gaps when multiple streaming pipelines run in the same JVM.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.