[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.

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR resolves a race condition in the Spark Runner DStream streaming mode where SingleEmitInputDStream could generate subsequent micro-batch RDDs before WatermarkAdvancingStreamingListener finished advancing watermarks for the completed batch. The changes cleanly synchronize SingleEmitInputDStream batch generation with GlobalWatermarkHolder in portable streaming pipelines, add missing thread synchronization to GlobalWatermarkHolder, and ensure static watermark and metrics state is cleared between pipeline runs.

🔍 General Feedback

• Gating watermark synchronization in SingleEmitInputDStream behind forceWatermarkSync and guarding against repeated compute calls for the same validTime (validTime.milliseconds() > lastValidTimeMs) safely aligns portable Impulse/Flatten streams with TestDStream without affecting existing callers.
• Synchronizing GlobalWatermarkHolder accessors (add, addAll, hasWatermarks, get) on the class monitor and resetting watermarkCache in clear() eliminates cross-thread visibility and state leakage issues across sequential runs.

---

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

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 59.26%. Comparing base (d7d31e9) to head (03c2a65).
⚠️ Report is 1 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff            @@
##             master   #41286   +/-   ##
=========================================
  Coverage     59.26%   59.26%           
  Complexity     4925     4925           
=========================================
  Files          1462     1462           
  Lines        186313   186313           
  Branches       3669     3669           
=========================================
  Hits         110417   110417           
  Misses        72540    72540           
  Partials       3356     3356           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.