damccorm opened a new pull request, #41286:
URL: https://github.com/apache/beam/pull/41286
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`](https://github.com/apache/beam/actions/runs/43388934435)
--
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]
