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:
@mxm Thanks for taking a look!
The problem isn't only that the future is not renewed after each completion
of a source. It also happens 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()` in the Hybrid Source `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]
