stevenzwu commented on code in PR #9308:
URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431856035
##########
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 wanted to add logging to the blocking path, but do not want to swamp the
log with a new message every second.
I think it is good to log this as TRACE only. we really need an INFO for the
first instance, because we already have an INFO log in `pauseOrResumeSplits`
method.
it seems that we might just need this while loop
```
while (pausedSplits.contains(currentSplitId)) {
LOG.trace("Reader {} is still paused reading split {}",
indexOfSubtask, currentSplitId);
synchronized (wakeUp) {
if (wakeUp.get()) {
wakeUp.set(false);
return new RecordsBySplits(Collections.emptyMap(),
Collections.emptySet());
}
try {
wakeUp.wait(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException("Interrupted while paused due to
watermark alignment.", e);
}
```
--
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]
