This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ea2eabe7da5 [FLINK-35214][runtime] Update result partition id for
remote input channel when unknown input channel is updated
ea2eabe7da5 is described below
commit ea2eabe7da555653ac946b185ed62a1d3bfd6c00
Author: Yuxin Tan <[email protected]>
AuthorDate: Tue Apr 23 13:47:22 2024 +0800
[FLINK-35214][runtime] Update result partition id for remote input channel
when unknown input channel is updated
---
.../partition/consumer/SingleInputGate.java | 3 +-
.../partition/consumer/UnknownInputChannel.java | 5 ++-
.../partition/consumer/SingleInputGateTest.java | 48 ++++++++++++++++++++++
3 files changed, 53 insertions(+), 3 deletions(-)
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 6a8dc3fe647..5fdbc6aeb78 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
@@ -652,7 +652,8 @@ public class SingleInputGate extends IndexedInputGate {
} else {
RemoteInputChannel remoteInputChannel =
unknownChannel.toRemoteInputChannel(
- shuffleDescriptor.getConnectionId());
+ shuffleDescriptor.getConnectionId(),
+
shuffleDescriptor.getResultPartitionID());
remoteInputChannel.setup();
newChannel = remoteInputChannel;
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java
index 0a5fa0dc8b9..6980ed507da 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java
@@ -168,11 +168,12 @@ class UnknownInputChannel extends InputChannel implements
ChannelStateHolder {
// Graduation to a local or remote input channel at runtime
// ------------------------------------------------------------------------
- public RemoteInputChannel toRemoteInputChannel(ConnectionID
producerAddress) {
+ public RemoteInputChannel toRemoteInputChannel(
+ ConnectionID producerAddress, ResultPartitionID resultPartitionID)
{
return new RemoteInputChannel(
inputGate,
getChannelIndex(),
- partitionId,
+ resultPartitionID,
consumedSubpartitionIndexSet,
checkNotNull(producerAddress),
connectionManager,
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
index 022cee17f23..4d74bf78f22 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
@@ -50,6 +50,7 @@ import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.buffer.TestingBufferPool;
import
org.apache.flink.runtime.io.network.partition.BufferAvailabilityListener;
import
org.apache.flink.runtime.io.network.partition.BufferWritingResultPartition;
import org.apache.flink.runtime.io.network.partition.ChannelStateHolder;
@@ -569,6 +570,53 @@ class SingleInputGateTest extends InputGateTestBase {
assertThat(newChannel.partitionId).isEqualTo(newPartitionId);
}
+ /**
+ * Test unknown input channel can set resultPartitionId correctly when
update to remote input
+ * channel, this occurs in the case of speculative execution that unknown
input channel only
+ * carries original resultPartitionId.
+ */
+ @Test
+ void testUpdateRemoteInputChannelWithNewPartitionId() throws Exception {
+ int bufferSize = 1024;
+ SingleInputGate inputGate = createInputGate(1);
+
+ TestingResultPartitionManager partitionManager =
+ new TestingResultPartitionManager(new
NoOpResultSubpartitionView());
+
+ ResultPartitionID oldPartitionId = new ResultPartitionID();
+
+ InputChannel unknown =
+ InputChannelBuilder.newBuilder()
+ .setPartitionManager(partitionManager)
+ .setPartitionId(oldPartitionId)
+ .buildUnknownChannel(inputGate);
+ inputGate.setInputChannels(unknown);
+
+ ResultPartitionID resultPartitionID = unknown.getPartitionId();
+ assertThat(resultPartitionID).isEqualTo(oldPartitionId);
+
+ ResultPartitionID newPartitionId =
+ new ResultPartitionID(
+ // speculative execution have the same
IntermediateResultPartitionID with
+ // original, only executionAttemptID is different.
+ oldPartitionId.getPartitionId(),
ExecutionAttemptID.randomId());
+ NettyShuffleDescriptor nettyShuffleDescriptor =
+ NettyShuffleDescriptorBuilder.newBuilder()
+ .setId(newPartitionId)
+ .setProducerLocation(ResourceID.generate())
+ .buildRemote();
+ inputGate.setBufferPool(
+ TestingBufferPool.builder()
+ .setRequestMemorySegmentSupplier(
+ () ->
MemorySegmentFactory.allocateUnpooledSegment(bufferSize))
+ .build());
+ inputGate.updateInputChannel(ResourceID.generate(),
nettyShuffleDescriptor);
+
+ InputChannel newChannel = inputGate.getChannel(0);
+ assertThat(newChannel).isInstanceOf(RemoteInputChannel.class);
+ assertThat(newChannel.partitionId).isEqualTo(newPartitionId);
+ }
+
/**
* Tests that the release of the input gate is noticed while polling the
channels for available
* data.
