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 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
##########
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()` 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]
