damccorm commented on code in PR #41013:
URL: https://github.com/apache/beam/pull/41013#discussion_r4556472965
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java:
##########
@@ -1012,6 +1012,22 @@ public UnboundedSourceRestriction<OutputT, CheckpointT>
currentRestriction() {
watermark);
}
+ /**
+ * Whether there is nothing left to read, like {@code
currentRestriction().getSource()
+ * instanceof EmptyUnboundedSource} but without calling {@link
+ * UnboundedReader#getCheckpointMark()}: a checkpoint ends the bundle of
elements read so far,
+ * so a checkpoint taken only to report progress would never be
committed nor finalized.
+ */
+ private boolean isDone() {
+ @Nullable UnboundedReader<OutputT> reader = currentReader;
+ if (reader == null || !readerHasBeenStarted) {
+ return initialRestriction.getSource() instanceof
EmptyUnboundedSource;
+ }
+ return reader instanceof EmptyUnboundedSource.EmptyUnboundedReader
+ || BoundedWindow.TIMESTAMP_MAX_VALUE.equals(
+ ensureTimestampWithinBounds(reader.getWatermark()));
+ }
Review Comment:
While removing `currentRestriction()` from `tryGetProgressOrThrow()`
eliminates the mid-bundle progress checkpoints,
[`UnboundedSourceAsSDFWrapperFn.processElement`](https://github.com/apache/beam/blob/d8084e89d334260719129740933c7af309187f09/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java#L609-L636)
still calls `currentRestriction()` -> `currentReader.getCheckpointMark()` **3
times** at the end of every bundle where `start()` or `advance()` returns
`false` (`out[0] == null`, so `currentReader` is not yet an
`EmptyUnboundedReader`):
1. **Line 610:** `UnboundedSourceRestriction<OutputT, CheckpointT>
currentRestriction = tracker.currentRestriction();` takes **Checkpoint 1** and
registers `checkpoint::finalizeCheckpoint` with
`bundleFinalizer.afterBundleCommit(...)`.
2. **Line 620:** `&& !(tracker.currentRestriction().getCheckpoint()
instanceof NoopCheckpointMark)` calls `tracker.currentRestriction()` a **second
time** instead of checking the local variable `!(checkpoint instanceof
NoopCheckpointMark)` from line 617, taking **Checkpoint 2** and immediately
dropping it.
3. **Line 636 (`ProcessContinuation.resume()`):** Returning
`ProcessContinuation.resume()` causes the runner/harness (`FnApiDoFnRunner` /
`OutputAndTimeBoundedSplittableProcessElementInvoker`) to immediately invoke
`tracker.trySplit(0)`, which at line 1035 calls `currentRestriction()` a
**third time** (**Checkpoint 3**).
Because Checkpoint 1 at line 610 already drained stateful checkpoint
preparers (like `JmsIO`, `MqttIO`, `SolaceIO`), Checkpoint 2 is dropped and
Checkpoint 3 (which is returned in `SplitResult` as the residual restriction to
the runner and used as the `cachedReaders` cache key!) is a different, empty
checkpoint whose `oldestMessageTimestamp` in `JmsIO`/`MqttIO` was reset to
`Instant.now()`.
In the core `Read.java` PR, could we at minimum replace
`!(tracker.currentRestriction().getCheckpoint() instanceof NoopCheckpointMark)`
at line 620 with `!(checkpoint instanceof NoopCheckpointMark)`, and ideally
avoid taking separate checkpoints at line 610 and inside `trySplit(0)` when the
bundle ends with `ProcessContinuation.resume()`?
--
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]
