Github user NicoK commented on a diff in the pull request:
https://github.com/apache/flink/pull/4559#discussion_r157548033
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
---
@@ -52,6 +54,10 @@
/** Flag indicating whether the subpartition has been released. */
private volatile boolean isReleased;
+ /** The number of non-event buffers currently in this subpartition */
+ @GuardedBy("buffers")
+ private volatile int buffersInBacklog;
--- End diff --
I shortly thought about relying on `buffers.size()` here to reduce
complexity and code, but `ArrayDeque#size()` (for `getBuffersInBacklog()`) may
show some race conditions then without synchronisation. However, if we picked
up the idea again of returning the backlog size with the buffer itself (which
is retrieved under the lock), i.e. similar to `BufferAndAvailability` being
returned by the `SequenceNumberingViewReader`, this would work and we would not
need the `volatile` here. Since you split the implementations into
`PipelinedSubpartition` and `SpillableSubpartition` anyway, this would be a
viable approach again.
What do you think? What would you prefer?
---
