This is an automated email from the ASF dual-hosted git repository. 1996fanrui pushed a commit to branch release-2.3 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0a2040c8085c8d34aa8982507da487a1cdaa0992 Author: Rui Fan <[email protected]> AuthorDate: Thu Jul 2 00:52:11 2026 +0200 [FLINK-40016][checkpoint] Refactor getBufferAndAvailability to return BufferAndAvailability directly --- .../partition/consumer/LocalInputChannel.java | 29 ++++++++++------------ 1 file changed, 13 insertions(+), 16 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java index 661e4b063c7..55ff29619c4 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java @@ -329,10 +329,10 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit seq++)); } - return getBufferAndAvailability(toBeConsumedBuffers.removeFirst()); + return Optional.of(getBufferAndAvailability(toBeConsumedBuffers.removeFirst())); } - return getBufferAndAvailability(next); + return Optional.of(getBufferAndAvailability(next)); } /** @@ -366,12 +366,13 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit } } - return getBufferAndAvailability( - new BufferAndBacklog( - next.buffer(), - next.buffersInBacklog(), - expectedNextDataType, - next.getSequenceNumber())); + return Optional.of( + getBufferAndAvailability( + new BufferAndBacklog( + next.buffer(), + next.buffersInBacklog(), + expectedNextDataType, + next.getSequenceNumber()))); } BufferAndBacklog next = toBeConsumedBuffers.removeFirst(); @@ -394,10 +395,10 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit next.getSequenceNumber()); } } - return getBufferAndAvailability(next); + return Optional.of(getBufferAndAvailability(next)); } - private Optional<BufferAndAvailability> getBufferAndAvailability(BufferAndBacklog next) + private BufferAndAvailability getBufferAndAvailability(BufferAndBacklog next) throws IOException { Buffer buffer = next.buffer(); if (buffer instanceof FileRegionBuffer) { @@ -419,12 +420,8 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit channelInfo, channelStatePersister, next.getSequenceNumber()); - return Optional.of( - new BufferAndAvailability( - buffer, - next.getNextDataType(), - next.buffersInBacklog(), - next.getSequenceNumber())); + return new BufferAndAvailability( + buffer, next.getNextDataType(), next.buffersInBacklog(), next.getSequenceNumber()); } @Override
