Github user NicoK commented on a diff in the pull request:
https://github.com/apache/flink/pull/4552#discussion_r161287607
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/SequenceNumberingViewReader.java
---
@@ -77,6 +92,46 @@ void requestSubpartitionView(
}
}
+ /**
+ * The credits from consumer are added in incremental way.
+ *
+ * @param creditDeltas The credit deltas
+ */
+ public void addCredit(int creditDeltas) {
+ numCreditsAvailable += creditDeltas;
+ }
+
+ /**
+ * Updates the value to indicate whether the reader is enqueued in the
pipeline or not.
+ *
+ * @param isRegisteredAvailable True if this reader is already enqueued
in the pipeline.
+ */
+ public void notifyAvailabilityChanged(boolean isRegisteredAvailable) {
+ this.isRegisteredAvailable = isRegisteredAvailable;
+ }
+
+ public boolean isRegisteredAvailable() {
+ return isRegisteredAvailable;
+ }
+
+ /**
+ * Check whether this reader is available or not.
+ *
+ * <p>Return true only if the next buffer is event or the reader has
both available
+ * credits and buffers.
+ */
+ public boolean isAvailable() {
+ if (numBuffersAvailable.get() <= 0) {
+ return false;
+ }
+
+ if (subpartitionView.nextBufferIsEvent() || numCreditsAvailable
> 0) {
--- End diff --
one more thing: in the special case of using it inside `getNextBuffer()`,
we already retrieved the `remaining` number of buffers so we can spare one
lookup into an atomic integer here by passing this one in
---
