Github user NicoK commented on a diff in the pull request: https://github.com/apache/flink/pull/4552#discussion_r161282684 --- 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 -- also, why not simply do the following for this whole method? ``` return numBuffersAvailable.get() > 0 && (numCreditsAvailable > 0 || subpartitionView.nextBufferIsEvent()); ```
---