damccorm commented on code in PR #41013:
URL: https://github.com/apache/beam/pull/41013#discussion_r4556472978
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java:
##########
@@ -311,12 +439,42 @@ JmsCheckpointMark newCheckpoint(
return checkpointMark;
}
+ /**
+ * A checkpoint without messages. Finalizing it still acknowledges the
earlier checkpoints of
+ * this reader.
+ */
JmsCheckpointMark emptyCheckpoint() {
- return new JmsCheckpointMark(oldestMessageTimestamp, null, null, null,
null);
+ return new JmsCheckpointMark(
+ pendingCheckpoints,
+ pendingCheckpoints.nextSequence(),
+ oldestMessageTimestamp,
+ null,
+ 0,
+ null,
+ null,
+ null);
}
boolean isEmpty() {
return messages.isEmpty();
Review Comment:
In `Preparer`, `add()`, `getOldestMessageTimestamp()`, `discard()`, and
`newCheckpoint()` all guard access to `messages` and `oldestMessageTimestamp`
with `lock.readLock()` / `lock.writeLock()`, whereas `emptyCheckpoint()` (which
reads `oldestMessageTimestamp` and calls `pendingCheckpoints.nextSequence()`)
and `isEmpty()` (which reads `messages.isEmpty()`) do not acquire
`lock.readLock()`.
Also, in `UnboundedJmsReader.takeCheckpoint()`, `if
(checkpointMarkPreparer.isEmpty())` is checked outside `synchronized (this)`
before entering `synchronized (this)` to call `newCheckpoint(...)`. Acquiring
`lock.readLock()` in `isEmpty()` and `emptyCheckpoint()` (and checking
`isEmpty()` inside `synchronized (this)` in `takeCheckpoint()`) would avoid any
race if `getCheckpointMark()` is called concurrently with `advance()`.
##########
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()`?
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java:
##########
@@ -633,6 +633,34 @@ public interface MessageMapper<T> extends Serializable {
T mapMessage(Message message) throws Exception;
}
+ /** The first stack frames outside JmsIO, to tell which runner code path
calls the reader. */
+ private static String callerOutsideJmsIO() {
+ String jmsPackage = JmsIO.class.getName().substring(0,
JmsIO.class.getName().lastIndexOf('.'));
+ StringBuilder caller = new StringBuilder();
+ int frames = 0;
+ for (StackTraceElement frame : new Throwable().getStackTrace()) {
+ if (frame.getClassName().startsWith(jmsPackage)) {
+ continue;
+ }
+ if (frames > 0) {
+ caller.append(" <- ");
+ }
+
caller.append(frame.getClassName()).append('.').append(frame.getMethodName());
+ if (++frames == 3) {
+ break;
+ }
+ }
+ return caller.toString();
Review Comment:
Nit: Since this only runs when `LOG.isDebugEnabled()`, `new
Throwable().getStackTrace()` is fine, though `StackWalker.getInstance()` avoids
materializing the full stack trace array if `DEBUG` logging is enabled on a
busy worker.
--
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]
