mas-chen commented on code in PR #20215:
URL: https://github.com/apache/flink/pull/20215#discussion_r945977650
##########
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:
Thanks for taking a look!
The problem isn't only that the future is not renewed after each completion.
It also because `whenComplete()` executes, the future is marked at complete and
is never renewed (for the underlying reader to tell to the runtime to stop
invoking pollNext()).
I think it can be refactored in other means. Proxying the
`currentReader.isAvailable()` could also work since the underlying reader
itself handles renewing the availability future. I'll take a look
--
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]
