pvary commented on code in PR #9308:
URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431255938
##########
flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java:
##########
@@ -80,6 +88,35 @@ public RecordsWithSplitIds<RecordAndPosition<T>> fetch()
throws IOException {
}
if (currentReader.hasNext()) {
+ if (pausedSplits.contains(currentSplitId)) {
+ // Wait until the reader is unblocked. Wake every second to catch any
missed signal.
+ // Return empty records if wakeUp is called, so pauseOrResumeSplits
could be processed.
+ boolean first = true;
Review Comment:
I am generally very-very cautious about using loop-less blocking calls.
Here is an example:
- WT (Watermark Thread) - Watermark arrives, and the decision is to block
this reader
- WT - `wakeup` is called
- ST (SplitReader Thread) - Current `fetch` finished
- ST - `pauseOrResumeSplits` is called
- ST - New `fetch` called
- ST - New `fetch` is checking if the split is paused, and find that we need
to block, but do not start to wait on the lock yet.
- WT - New watermark arrives, and the decision is to unblock the reader
- WT - `wakeup` is called
- ST - If the `fetch` arrives to the locking code now, then it will not
receive the `wakeup` call
The main issue is that the `wakeUp` and the `pauseOrResumeSplits` is not
atomic, we need another flag to store if the given `wakeUp` is handled, or not.
So i think we minimally need to have something like this:
```
if (pausedSplits.contains(currentSplitId)) {
LOG.info("Reader {} paused reading split {}", indexOfSubtask,
currentSplitId);
synchronized (wakeUp) {
if (wakeUp.get()) {
wakeUp.set(false);
return new RecordsBySplits(Collections.emptyMap(),
Collections.emptySet());
}
try {
wakeUp.wait();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Interrupted while blocked on
reading.", e);
}
}
}
}
```
I think compared to this the added security of the `while` loop is a good
compromise.
WDYT?
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
