pvary commented on code in PR #9308:
URL: https://github.com/apache/iceberg/pull/9308#discussion_r1431841694
##########
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:
The timeline is only valid if we do not use an extra flag to store the
`wakeUp` state and rely on a simple `wait-notify` pattern.
I used this timeline to describe why I introduced the `wakeUp` flag, and use
synchronize around it in the PR.
I make sure that the `wakeUp` flag is only read/set inside a `synchronized`
block, and we return from the `fetch` while holding the synchronization lock.
So if there is a concurrent `wakeUp` method call, it will be blocked until we
return, and we handle the next wakeup in the next fetch loop
--
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: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
