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 this additional complexity)
Also in the code snippet above, I think there could be a need for
synchronization on the future, 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
