zhijiangW opened a new pull request #7911: [FLINK-11082][network] Fix the logic of getting backlog in sub partition URL: https://github.com/apache/flink/pull/7911 ## What is the purpose of the change *In current implementation, the backlog would be increased by 1 once `BufferConsumer` adding into sub partition, and decreased by 1 once removing from the sub partition. It is not reasonable in some scenarios. For example, there are less data in some sub partitions which results in the `BufferConsumer` is not finished long time, especially for no flush mechanism in batch job, that means this `BufferConsumer` can not be transferred in network to be consumed by downstream side. But this `BufferConsumer` is already calculated in the backlog, then the downstream would request one more floating buffer for this backlog, but this floating buffer actually not used for long time. So the floating buffer usage is not reasonable in this case. * *We could still retain the previous logic for increasing and decreasing backlog during enqueue and dequeue. But the logic of getting backlog is adjusted by not only relying the number of non-event buffers in sub partition, but also considering the conditions of `flush` or `finish` status. If the sub partition is flushed or finished, the backlog should be exactly the same with the number of non-event buffers, otherwise we could regard the last `BufferConsumer` in unfinished state to decrease the current value by 1.* ## Brief change log - *Define abstract `getBuffersInBacklog` in `ResultSubpartition`* - *Implement this new abstract method in `PipelinedSubpartition` and `SpillableSubpartition` * - *Return `void` in `decreaseBacklog`* ## Verifying this change *This change is already covered by existing tests, such as `PipelinedSubpartitionWithReadViewTest` and `SpillableSubpartitionTest`.* *Add one more new test `testBacklogWithFinishedPartition` in `PipelinedSubpartitionWithReadViewTest`.* ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (yes / **no**) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**) - The serializers: (yes / **no** / don't know) - The runtime per-record code paths (performance sensitive): (yes / no / **don't know**) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / **no** / don't know) - The S3 file system connector: (yes / **no** / don't know) ## Documentation - Does this pull request introduce a new feature? (yes / **no**) - If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
