[Spark] Retain the last watermark of a source with no update in the batch

[Eliaaazzz]

GlobalWatermarkHolder.advance() rebuilt the global watermark map from only the sources that had a queued update for the completed batch and overwrote the stored map with the result. A source that reported no progress in that batch lost its last known watermark, and since SparkTimerInternals.forStreamFromSources skips ids that are missing from the map, the input watermark of every consuming stage jumped ahead of the idle source, firing timers early and dropping its later elements as late.

computeNewWatermarks now starts from the stored map and merges the per-source updates over it, so an idle source keeps its last watermark until it reports again. A source whose high watermark reached the end of time is done and holds nothing back, so it is not retained past the batch that completed it; without this, its finite synchronized processing time and low watermark would constrain consumers forever. Retained sources can disagree on the synchronized processing time, so SparkTimerInternals takes the slowest one instead of asserting equality. The stored map is also read once per advance() instead of once per updated source.

Two CreateStream test scripts (testFlattenedWithWatermarkHold, shouldTriggerProcessingTimeTimerWithSparseKey) now advance their watermarks to infinity like the other scripts. Their scripts previously stopped at a finite watermark and the tests only terminated because the dropped source no longer held the global watermark back; with retention that stalled watermark would legitimately hold it back until the test timeout.

testWatermarkRetainedForSourceWithoutUpdate fails on master with source 2 absent from the map after the second advance(); testCompletedSourceAgesOut pins the report-then-age-out behavior.

Fixes #39822.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[github-actions]

Assigning reviewers:

R: @damccorm added as fallback since no labels match configuration

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[Abacn]

cc: @twosom @kennknowles who had discussion in #23129

I believe this is one deterministic cause of the watermark races described in #23129.

Have you verified if any validate runner tests previously failing now would be fixed by this change?

[Eliaaazzz]

Good question, I dug into this. I checked and could not find a ValidatesRunner test that was failing before and turns green with this fix, and I think there is a structural reason.

The recent runs of both Spark postcommits are green on master, so I focused on the excluded tests. I lifted the UsesTestStreamWithMultipleStages exclusion locally and ran TestStreamTest on this branch. testMultiStage still fails at pipeline translation with "EVENT_TIME not yet supported in streaming mode" from StatefulStreamingParDoEvaluator, so its blocker is stateful event time timer support and it never reaches the watermark code.

The other five tests selected by that run pass, and the full enabled validatesRunnerStreaming suite is green on this branch locally, 284 tests, no failures, one skipped.

The drop needs at least two source ids, one updating in a batch while another already registered source with an unfinished watermark stays idle. A TestStream reports its watermark only on watermark events, and the only VR test I found with two TestStreams, testMultipleStreams, advances both to infinity together, so it does not exercise this case. It passes before and after this change.

That is why the direct regression coverage here is the new GlobalWatermarkHolderTest case. Two existing CreateStream scripts relied on the drop erasing their stalled source, with retention they now advance their watermarks to infinity explicitly like the other scripts.

If a VR level regression test would be useful I can add one with two TestStreams, one advancing while the other holds its watermark after registering. I would verify it fails on master before pushing it.

I will resolve the conflict from the #39825 merge and trigger both Spark postcommits on this PR.

[github-actions]

Reminder, please take a look at this pr: @damccorm

[damccorm]

Thanks, this generally LGTM, just had one question

[damccorm]

How is this used/why do we only return if there are updates? Does it actually save us anything?

[Eliaaazzz]

hasUpdates restores a signal the pre-change code got for free.

The caller is advance(batchId), which skips both publication calls when the returned map is empty:

final Map<Integer, SparkWatermarks> newWatermarks = computeNewWatermarks(blockManager);
if (!newWatermarks.isEmpty()) {
  writeRemoteWatermarkBlock(newWatermarks, blockManager);
  writeLocalWatermarkCopy(newWatermarks);
} else {
  LOG.info("No new watermarks could be computed upon completion of batch: {}", batchId);
}

Before this change newValues started empty and only a source with a queued update got an entry, so a batch where every queue was empty produced an empty map and advance published nothing. Seeding newValues from the stored map means a quiet batch now yields a non-empty map whenever a stored source still has high < TIMESTAMP_MAX_VALUE. hasUpdates is what keeps that batch from publishing.

What it saves on such a batch is the removeBlock plus putSingle pair and the local copy assignment. The comments on writeRemoteWatermarkBlock, and #18426 which they link, describe the window where an executor fetching WATERMARKS_BLOCK_ID finds nothing. The saving is narrow: the block fetch, the seeding and the queue scan all still happen.

One consequence is worth stating plainly. On a batch where every queue is empty, the stored block keeps a source whose high watermark already reached TIMESTAMP_MAX_VALUE, which seeding would have pruned. Such a source can still hold the effective low watermark back, because SparkTimerInternals.forStreamFromSources takes the slowest low watermark and a completed high watermark does not imply a completed low watermark. testCompletedSourceAgesOut uses exactly that shape, low at +5ms with high at TIMESTAMP_MAX_VALUE. The pre-change code also published nothing on an all-quiet batch, so that persistence predates this PR. What the guard defers is the pruning seeding introduced. A completed source drops out on the next published batch where its own queue is empty and another source supplies an update, and the update loop puts it back on any batch where it does report.

Dropping the flag is not the same as publishing every batch. On a quiet batch it would publish the seeded map, which prunes completed sources as long as one unfinished source remains. Once every stored source has completed, seeding yields an empty map and advance publishes nothing regardless, so clearing that block would need advance to tell "nothing to publish" apart from "publish this empty map".

The tests cover retention and pruning on a batch where another source updates. They do not cover the all-queues-empty path. I can add that case if you want the behavior pinned down.

[damccorm]

Thanks, this helps

[damccorm]

Thanks!

[damccorm]

Thanks, this helps