gaoyunhaii commented on a change in pull request #16366:
URL: https://github.com/apache/flink/pull/16366#discussion_r665140835
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
##########
@@ -350,10 +350,21 @@ BufferAndBacklog pollBuffer() {
}
void resumeConsumption() {
+ final boolean notifyDataAvailable;
synchronized (buffers) {
checkState(isBlocked, "Should be blocked by checkpoint.");
isBlocked = false;
+
+ // When buffer timeout is 0, there might be data produced during
+ // the channel is blocking and do not notify due to blocking, and
+ // it would not get chance to notify, thus we should notify for it
+ // here.
+ notifyDataAvailable = isDataAvailableUnsafe();
+ }
+
+ if (notifyDataAvailable) {
+ notifyDataAvailable();
Review comment:
Very thanks pointing out this issue! Sorry that I ignored the logic
corresponding to the `RemoteInputChannel`. I think it is reasonable to not
introduce the duplicate notification, I'll modify the code.
--
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]
