This is an automated email from the ASF dual-hosted git repository. 1996fanrui pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit ed2a5d4050ca2c269092388c8c89884b5634d41a Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 00:03:40 2026 +0200 [FLINK-39520][network] Add InputGate#getChannel(InputChannelInfo) - InputGate: new abstract InputChannel getChannel(InputChannelInfo). - SingleInputGate: channels[channelInfo.getInputChannelIdx()]. UnionInputGate: resolve via inputGatesByGateIndex.get(channelInfo.getGateIdx()) .getChannel(channelInfo) (correct global-vs-member index semantics). InputGateWithMetrics: delegate. CheckpointedInputGate: delegate helper. - AbstractStreamTaskNetworkInput: getChannel(channelInfo.getInputChannelIdx()) -> getChannel(channelInfo). - Test impls: MockInputGate, MockIndexedInputGate, AlignedCheckpointsMassiveRandomTest's inner gate. --- .../flink/runtime/io/network/partition/consumer/InputGate.java | 8 ++++++++ .../runtime/io/network/partition/consumer/SingleInputGate.java | 5 +++++ .../runtime/io/network/partition/consumer/UnionInputGate.java | 7 +++++++ .../apache/flink/runtime/taskmanager/InputGateWithMetrics.java | 5 +++++ .../streaming/runtime/io/AbstractStreamTaskNetworkInput.java | 3 +-- .../streaming/runtime/io/checkpointing/CheckpointedInputGate.java | 4 ++++ .../apache/flink/streaming/runtime/io/MockIndexedInputGate.java | 5 +++++ .../java/org/apache/flink/streaming/runtime/io/MockInputGate.java | 5 +++++ .../io/checkpointing/AlignedCheckpointsMassiveRandomTest.java | 5 +++++ 9 files changed, 45 insertions(+), 2 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java index dd744bae330..8c5b02fb565 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java @@ -141,6 +141,14 @@ public abstract class InputGate /** Returns the channel of this gate. */ public abstract InputChannel getChannel(int channelIndex); + /** + * Returns the channel identified by {@code channelInfo}. Unlike {@link #getChannel(int)}, whose + * index is gate-global, this resolves through the full {@code (gateIdx, inputChannelIdx)} pair, + * so it stays correct for {@link UnionInputGate} where the global index differs from a member + * gate's local channel index. + */ + public abstract InputChannel getChannel(InputChannelInfo channelInfo); + /** Returns the channel infos of this gate. */ public List<InputChannelInfo> getChannelInfos() { return IntStream.range(0, getNumberOfInputChannels()) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java index 438efa2f58b..1e6b8168f34 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java @@ -595,6 +595,11 @@ public class SingleInputGate extends IndexedInputGate { return channels[channelIndex]; } + @Override + public InputChannel getChannel(InputChannelInfo channelInfo) { + return channels[channelInfo.getInputChannelIdx()]; + } + // ------------------------------------------------------------------------ // Setup/Life-cycle // ------------------------------------------------------------------------ diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java index dda71c63be3..8d1ad255377 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java @@ -176,6 +176,13 @@ public class UnionInputGate extends InputGate { .getChannel(channelIndex - inputGateChannelIndexOffsets[gateIndex]); } + @Override + public InputChannel getChannel(InputChannelInfo channelInfo) { + // The member gate's local channel index is carried directly in channelInfo, so resolve + // through the owning gate instead of the gate-global getChannel(int) addressing. + return inputGatesByGateIndex.get(channelInfo.getGateIdx()).getChannel(channelInfo); + } + @Override public boolean isFinished() { return inputGatesWithRemainingData.isEmpty(); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java index bff412f53b3..3eff82a6eb9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java @@ -80,6 +80,11 @@ public class InputGateWithMetrics extends IndexedInputGate { return inputGate.getChannel(channelIndex); } + @Override + public InputChannel getChannel(InputChannelInfo channelInfo) { + return inputGate.getChannel(channelInfo); + } + @Override public int getGateIndex() { return inputGate.getGateIndex(); diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java index a3743edec7f..26f0b5f8dff 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java @@ -326,8 +326,7 @@ public abstract class AbstractStreamTaskNetworkInput< // terminate the TM Exception err = null; for (InputChannelInfo channelInfo : new ArrayList<>(recordDeserializers.keySet())) { - final boolean hadError = - checkpointedInputGate.getChannel(channelInfo.getInputChannelIdx()).hasError(); + final boolean hadError = checkpointedInputGate.getChannel(channelInfo).hasError(); try { releaseDeserializer(channelInfo); } catch (Exception e) { diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointedInputGate.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointedInputGate.java index ad0dcb5b0bb..ec4cde4de58 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointedInputGate.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointedInputGate.java @@ -296,6 +296,10 @@ public class CheckpointedInputGate implements PullingAsyncDataInput<BufferOrEven return inputGate.getChannel(channelIndex); } + public InputChannel getChannel(InputChannelInfo channelInfo) { + return inputGate.getChannel(channelInfo); + } + public List<InputChannelInfo> getChannelInfos() { return inputGate.getChannelInfos(); } diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockIndexedInputGate.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockIndexedInputGate.java index 584aeb7eb90..5cc44377707 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockIndexedInputGate.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockIndexedInputGate.java @@ -86,6 +86,11 @@ public class MockIndexedInputGate extends IndexedInputGate { throw new UnsupportedOperationException(); } + @Override + public InputChannel getChannel(InputChannelInfo channelInfo) { + throw new UnsupportedOperationException(); + } + @Override public void setChannelStateWriter(ChannelStateWriter channelStateWriter) {} diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockInputGate.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockInputGate.java index 71b2c43f330..e63a49be53b 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockInputGate.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockInputGate.java @@ -101,6 +101,11 @@ public class MockInputGate extends IndexedInputGate { throw new UnsupportedOperationException(); } + @Override + public InputChannel getChannel(InputChannelInfo channelInfo) { + throw new UnsupportedOperationException(); + } + @Override public List<InputChannelInfo> getChannelInfos() { return IntStream.range(0, numberOfChannels) diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlignedCheckpointsMassiveRandomTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlignedCheckpointsMassiveRandomTest.java index 619873c387d..e27f6b28937 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlignedCheckpointsMassiveRandomTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlignedCheckpointsMassiveRandomTest.java @@ -178,6 +178,11 @@ class AlignedCheckpointsMassiveRandomTest { throw new UnsupportedOperationException(); } + @Override + public InputChannel getChannel(InputChannelInfo channelInfo) { + throw new UnsupportedOperationException(); + } + @Override public List<InputChannelInfo> getChannelInfos() { return IntStream.range(0, numberOfChannels)
