wsry commented on a change in pull request #16366:
URL: https://github.com/apache/flink/pull/16366#discussion_r664366954
##########
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:
For RemoteInputChannel, the available reader should have been enqueued
as available when resuming consumption in
PartitionRequestQueue#addCreditOrResumeConsumption method, this data available
notification can be duplicated. Maybe we can keep the logic of
LocalInputChannel#resumeConsumption method unchanged and only fix the
flushRequested flag? What do you think?
--
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]
