mas-chen commented on code in PR #20215:
URL: https://github.com/apache/flink/pull/20215#discussion_r947506841
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceReader.java:
##########
@@ -231,16 +236,7 @@ private void setCurrentReader(int index) {
reader.start();
currentSourceIndex = index;
currentReader = reader;
- currentReader
- .isAvailable()
- .whenComplete(
- (result, ex) -> {
- if (ex == null) {
- availabilityFuture.complete(result);
- } else {
- availabilityFuture.completeExceptionally(ex);
- }
- });
Review Comment:
If I understand the process in the alternative solution:
1. A new reader is set.
2. When the reader future is done, renew the future.
What is the process that renews the renewed future in step 2? Such a process
would need to be recursively ran outside of this reader switch handling method.
And repeatedly signaling completeness from the underlying reader and renewing
again and again. (This is why PR introduces the
MultipleFuturesAvailabilityHelper to coordinate this additional complexity)
Also in the code snippet above, I think there could be a need for
synchronization on the future variable, since `whenComplete()` would not be
guaranteed to run on the task thread, unless we use `whenCompleteAsync()` and
pass the task executor, which we don't have access to.
--
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...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
