Eliaaazzz commented on code in PR #39823:
URL: https://github.com/apache/beam/pull/39823#discussion_r3999156878
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/util/GlobalWatermarkHolder.java:
##########
@@ -240,7 +252,7 @@ private static Map<Integer, SparkWatermarks>
computeNewWatermarks(BlockManager b
new SparkWatermarks(nextLowWatermark, nextHighWatermark,
nextSynchronizedProcessingTime));
}
- return newValues;
+ return hasUpdates ? newValues : new HashMap<>();
Review Comment:
`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:
```java
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.
--
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]
