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 56dfbeb423a9884b4def84fa155ac21fc4e4a30e Author: Rui Fan <[email protected]> AuthorDate: Fri Oct 2 23:43:42 2026 +0200 [FLINK-40904][runtime] Pass consumer parallelism to tasks via ResultPartitionDeploymentDescriptor A task can only see the number of subpartitions of its outputs, which is not the consumer parallelism on a POINTWISE edge. Pass the parallelism of the consuming job vertices through ResultPartitionDeploymentDescriptor and expose it via Environment#getWriterConsumerParallelism. No behavior change. --- .../state/api/runtime/SavepointEnvironment.java | 5 ++++ .../ResultPartitionDeploymentDescriptor.java | 30 +++++++++++++++++++++- .../flink/runtime/execution/Environment.java | 8 ++++++ .../flink/runtime/executiongraph/Execution.java | 5 +++- .../runtime/taskmanager/RuntimeEnvironment.java | 10 ++++++++ .../org/apache/flink/runtime/taskmanager/Task.java | 8 ++++++ .../ResultPartitionDeploymentDescriptorTest.java | 8 +++++- .../runtime/deployment/ShuffleDescriptorTest.java | 3 ++- .../partition/AbstractPartitionTrackerTest.java | 1 + .../io/network/partition/PartitionTestUtils.java | 3 ++- .../partition/ResultPartitionFactoryTest.java | 1 + .../operators/testutils/DummyEnvironment.java | 5 ++++ .../operators/testutils/MockEnvironment.java | 6 +++++ .../runtime/shuffle/NettyShuffleUtilsTest.java | 3 ++- .../taskexecutor/TaskExecutorSubmissionTest.java | 5 ++-- .../apache/flink/runtime/taskmanager/TaskTest.java | 3 ++- .../runtime/tasks/StreamMockEnvironment.java | 13 ++++++++++ .../streaming/runtime/tasks/StreamTaskITCase.java | 1 + 18 files changed, 109 insertions(+), 9 deletions(-) diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java index 52be03cde86..5bb7a60cdfa 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java @@ -328,6 +328,11 @@ public class SavepointEnvironment implements Environment { throw new UnsupportedOperationException(ERROR_MSG); } + @Override + public int getWriterConsumerParallelism(int index) { + throw new UnsupportedOperationException(ERROR_MSG); + } + @Override public ResultPartitionWriter[] getAllWriters() { return new ResultPartitionWriter[0]; diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java index 6cca01c788f..79b3910f007 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java @@ -28,6 +28,7 @@ import org.apache.flink.runtime.state.KeyGroupRangeAssignment; import java.io.Serializable; +import static org.apache.flink.util.Preconditions.checkArgument; import static org.apache.flink.util.Preconditions.checkNotNull; /** @@ -45,14 +46,31 @@ public class ResultPartitionDeploymentDescriptor implements Serializable { private final int maxParallelism; + /** Parallelism of the consuming job vertices, or {@link #UNKNOWN_CONSUMER_PARALLELISM}. */ + private final int consumerParallelism; + + /** + * The consumer parallelism is not decided yet when the producer is deployed. This only happens + * in dynamic graphs (AdaptiveBatchScheduler), where the parallelism of a consumer vertex may be + * decided after its producers are deployed. Every other partition carries the actual consumer + * parallelism. + */ + public static final int UNKNOWN_CONSUMER_PARALLELISM = -1; + public ResultPartitionDeploymentDescriptor( PartitionDescriptor partitionDescriptor, ShuffleDescriptor shuffleDescriptor, - int maxParallelism) { + int maxParallelism, + int consumerParallelism) { this.partitionDescriptor = checkNotNull(partitionDescriptor); this.shuffleDescriptor = checkNotNull(shuffleDescriptor); KeyGroupRangeAssignment.checkParallelismPreconditions(maxParallelism); this.maxParallelism = maxParallelism; + checkArgument( + consumerParallelism > 0 || consumerParallelism == UNKNOWN_CONSUMER_PARALLELISM, + "Invalid consumer parallelism %s.", + consumerParallelism); + this.consumerParallelism = consumerParallelism; } public IntermediateDataSetID getResultId() { @@ -88,6 +106,16 @@ public class ResultPartitionDeploymentDescriptor implements Serializable { return maxParallelism; } + /** + * Returns the parallelism of the job vertices consuming this partition, or {@link + * #UNKNOWN_CONSUMER_PARALLELISM} if it is not decided yet. Unlike {@link + * #getNumberOfSubpartitions()}, this is the actual consumer parallelism for every distribution + * pattern. + */ + public int getConsumerParallelism() { + return consumerParallelism; + } + public ShuffleDescriptor getShuffleDescriptor() { return shuffleDescriptor; } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java index 4b449a4d129..d989f674774 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java @@ -32,6 +32,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointMetrics; import org.apache.flink.runtime.checkpoint.TaskStateSnapshot; import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriteRequestExecutorFactory; import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter; +import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; import org.apache.flink.runtime.externalresource.ExternalResourceInfoProvider; import org.apache.flink.runtime.io.disk.iomanager.IOManager; @@ -248,6 +249,13 @@ public interface Environment { ResultPartitionWriter getWriter(int index); + /** + * Returns the parallelism of the job vertices consuming the partition written by {@link + * #getWriter(int)}, or {@link ResultPartitionDeploymentDescriptor#UNKNOWN_CONSUMER_PARALLELISM} + * if it is not decided yet. + */ + int getWriterConsumerParallelism(int index); + ResultPartitionWriter[] getAllWriters(); IndexedInputGate getInputGate(int index); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java index 46d5a8d60b0..95d9610e3fb 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java @@ -561,7 +561,10 @@ public class Execution IntermediateResultPartition partition, ShuffleDescriptor shuffleDescriptor) { return new ResultPartitionDeploymentDescriptor( - partitionDescriptor, shuffleDescriptor, getPartitionMaxParallelism(partition)); + partitionDescriptor, + shuffleDescriptor, + getPartitionMaxParallelism(partition), + partition.getIntermediateResult().getConsumersParallelism()); } /** diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java index 69df0116802..131af5bc335 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java @@ -58,6 +58,7 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; +import static org.apache.flink.util.Preconditions.checkArgument; import static org.apache.flink.util.Preconditions.checkNotNull; import static org.apache.flink.util.Preconditions.checkState; @@ -93,6 +94,7 @@ public class RuntimeEnvironment implements Environment { private final Map<String, Future<Path>> distCacheEntries; private final ResultPartitionWriter[] writers; + private final int[] writerConsumerParallelisms; private final IndexedInputGate[] inputGates; private final TaskEventDispatcher taskEventDispatcher; @@ -145,6 +147,7 @@ public class RuntimeEnvironment implements Environment { InputSplitProvider splitProvider, Map<String, Future<Path>> distCacheEntries, ResultPartitionWriter[] writers, + int[] writerConsumerParallelisms, IndexedInputGate[] inputGates, TaskEventDispatcher taskEventDispatcher, CheckpointResponder checkpointResponder, @@ -177,6 +180,8 @@ public class RuntimeEnvironment implements Environment { this.splitProvider = checkNotNull(splitProvider); this.distCacheEntries = checkNotNull(distCacheEntries); this.writers = checkNotNull(writers); + this.writerConsumerParallelisms = checkNotNull(writerConsumerParallelisms); + checkArgument(writers.length == writerConsumerParallelisms.length); this.inputGates = checkNotNull(inputGates); this.taskEventDispatcher = checkNotNull(taskEventDispatcher); this.checkpointResponder = checkNotNull(checkpointResponder); @@ -306,6 +311,11 @@ public class RuntimeEnvironment implements Environment { return writers[index]; } + @Override + public int getWriterConsumerParallelism(int index) { + return writerConsumerParallelisms[index]; + } + @Override public ResultPartitionWriter[] getAllWriters() { return writers; diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java index 3ef341c1372..52e01e59f96 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java @@ -230,6 +230,9 @@ public class Task private final ResultPartitionWriter[] partitionWriters; + /** Consumer parallelism of each partition in {@link #partitionWriters}, in the same order. */ + private final int[] partitionConsumerParallelisms; + private final IndexedInputGate[] inputGates; /** Connection to the task manager. */ @@ -423,6 +426,10 @@ public class Task .toArray(new ResultPartitionWriter[] {}); this.partitionWriters = resultPartitionWriters; + this.partitionConsumerParallelisms = + resultPartitionDeploymentDescriptors.stream() + .mapToInt(ResultPartitionDeploymentDescriptor::getConsumerParallelism) + .toArray(); // consumed intermediate result partitions final IndexedInputGate[] gates = @@ -731,6 +738,7 @@ public class Task inputSplitProvider, distributedCacheEntries, partitionWriters, + partitionConsumerParallelisms, inputGates, taskEventDispatcher, checkpointResponder, diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java index eddb41d6c78..181f807fff8 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java @@ -51,6 +51,8 @@ class ResultPartitionDeploymentDescriptorTest { private static final ResultPartitionType partitionType = ResultPartitionType.PIPELINED; private static final int numberOfSubpartitions = 24; + + private static final int consumerParallelism = 12; private static final int connectionIndex = 10; private static final boolean isBroadcast = false; private static final boolean isAllToAllDistribution = true; @@ -112,7 +114,10 @@ class ResultPartitionDeploymentDescriptorTest { ShuffleDescriptor shuffleDescriptor) throws IOException { ResultPartitionDeploymentDescriptor orig = new ResultPartitionDeploymentDescriptor( - partitionDescriptor, shuffleDescriptor, numberOfSubpartitions); + partitionDescriptor, + shuffleDescriptor, + numberOfSubpartitions, + consumerParallelism); ResultPartitionDeploymentDescriptor copy = CommonTestUtils.createCopySerializable(orig); verifyResultPartitionDeploymentDescriptorCopy(copy); return copy; @@ -125,5 +130,6 @@ class ResultPartitionDeploymentDescriptorTest { assertThat(partitionId).isEqualTo(copy.getPartitionId()); assertThat(partitionType).isEqualTo(copy.getPartitionType()); assertThat(numberOfSubpartitions).isEqualTo(copy.getNumberOfSubpartitions()); + assertThat(consumerParallelism).isEqualTo(copy.getConsumerParallelism()); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ShuffleDescriptorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ShuffleDescriptorTest.java index df4291170ca..79cae3fb573 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ShuffleDescriptorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ShuffleDescriptorTest.java @@ -274,6 +274,7 @@ class ShuffleDescriptorTest { .registerPartitionWithProducer( jobID, partitionDescriptor, producerDescriptor) .get(); - return new ResultPartitionDeploymentDescriptor(partitionDescriptor, shuffleDescriptor, 1); + return new ResultPartitionDeploymentDescriptor( + partitionDescriptor, shuffleDescriptor, 1, 1); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AbstractPartitionTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AbstractPartitionTrackerTest.java index 6f90313cea8..a49075b76db 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AbstractPartitionTrackerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AbstractPartitionTrackerTest.java @@ -87,6 +87,7 @@ public class AbstractPartitionTrackerTest { : Optional.empty(); } }, + 1, 1); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java index 20547adc2c7..1e921f5b3bf 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java @@ -137,7 +137,8 @@ public enum PartitionTestUtils { .setPartitionId(shuffleDescriptor.getResultPartitionID().getPartitionId()) .setPartitionType(partitionType) .build(); - return new ResultPartitionDeploymentDescriptor(partitionDescriptor, shuffleDescriptor, 1); + return new ResultPartitionDeploymentDescriptor( + partitionDescriptor, shuffleDescriptor, 1, 1); } public static PartitionedFile createPartitionedFile( diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java index bd8156634df..95907f5ab54 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java @@ -193,6 +193,7 @@ class ResultPartitionFactoryTest { .setIsBroadcast(isBroadcast) .build(), NettyShuffleDescriptorBuilder.newBuilder().buildLocal(), + 1, 1); // guard our test assumptions diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java index fe291c1b4e5..367b28d3d30 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java @@ -264,6 +264,11 @@ public class DummyEnvironment implements Environment { return null; } + @Override + public int getWriterConsumerParallelism(int index) { + return taskInfo.getNumberOfParallelSubtasks(); + } + @Override public ResultPartitionWriter[] getAllWriters() { return new ResultPartitionWriter[0]; diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java index 101ff4386cb..84114ee9ee1 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java @@ -332,6 +332,12 @@ public class MockEnvironment implements Environment, AutoCloseable { return outputs.get(index); } + /** Outputs are consumed with the same parallelism as this task. */ + @Override + public int getWriterConsumerParallelism(int index) { + return taskInfo.getNumberOfParallelSubtasks(); + } + @Override public ResultPartitionWriter[] getAllWriters() { return outputs.toArray(new ResultPartitionWriter[outputs.size()]); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java index 76f3bf38341..575d7addf23 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java @@ -194,7 +194,8 @@ class NettyShuffleUtilsTest { true, false); ResultPartitionDeploymentDescriptor resultPartitionDeploymentDescriptor = - new ResultPartitionDeploymentDescriptor(partitionDescriptor, shuffleDescriptor, 1); + new ResultPartitionDeploymentDescriptor( + partitionDescriptor, shuffleDescriptor, 1, 1); ExecutionAttemptID consumerID = createExecutionAttemptId(); Collection<ResultPartition> resultPartitions = diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java index 8e87b86fcad..65c37643f86 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java @@ -503,7 +503,7 @@ class TaskExecutorSubmissionTest { .build(); ResultPartitionDeploymentDescriptor resultPartitionDeploymentDescriptor = - new ResultPartitionDeploymentDescriptor(partitionDescriptor, sdd, 1); + new ResultPartitionDeploymentDescriptor(partitionDescriptor, sdd, 1, 1); TaskDeploymentDescriptor tdd = createTestTaskDeploymentDescriptor( "task", @@ -798,7 +798,8 @@ class TaskExecutorSubmissionTest { .setPartitionId(shuffleDescriptor.getResultPartitionID().getPartitionId()) .build(); ResultPartitionDeploymentDescriptor resultPartitionDeploymentDescriptor = - new ResultPartitionDeploymentDescriptor(partitionDescriptor, shuffleDescriptor, 1); + new ResultPartitionDeploymentDescriptor( + partitionDescriptor, shuffleDescriptor, 1, 1); return createTestTaskDeploymentDescriptor( "Sender", shuffleDescriptor.getResultPartitionID().getProducerId(), diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java index 3b326ba2946..c87d8915f0f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java @@ -309,7 +309,8 @@ class TaskTest { final ShuffleDescriptor shuffleDescriptor = NettyShuffleDescriptorBuilder.newBuilder().buildLocal(); final ResultPartitionDeploymentDescriptor dummyPartition = - new ResultPartitionDeploymentDescriptor(partitionDescriptor, shuffleDescriptor, 1); + new ResultPartitionDeploymentDescriptor( + partitionDescriptor, shuffleDescriptor, 1, 1); testExecutionFailsInNetworkRegistration( Collections.singletonList(dummyPartition), Collections.emptyList()); } diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java index dd4ef8994c2..27f37f24cf7 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java @@ -108,6 +108,9 @@ public class StreamMockEnvironment implements Environment { private List<ResultPartitionWriter> outputs; + /** Consumer parallelism of all outputs, equal to the parallelism of this task by default. */ + private int writerConsumerParallelism; + private final ExecutionAttemptID executionAttemptID; private final BroadcastVariableManager bcVarManager = new BroadcastVariableManager(); @@ -193,6 +196,7 @@ public class StreamMockEnvironment implements Environment { this.taskConfiguration = taskConfig; this.inputs = new LinkedList<>(); this.outputs = new LinkedList<ResultPartitionWriter>(); + this.writerConsumerParallelism = taskInfo.getNumberOfParallelSubtasks(); this.memManager = MemoryManagerBuilder.newBuilder().setMemorySize(offHeapMemorySize).build(); this.sharedResources = new SharedResources(); @@ -324,6 +328,15 @@ public class StreamMockEnvironment implements Environment { return outputs.get(index); } + @Override + public int getWriterConsumerParallelism(int index) { + return writerConsumerParallelism; + } + + public void setWriterConsumerParallelism(int writerConsumerParallelism) { + this.writerConsumerParallelism = writerConsumerParallelism; + } + @Override public ResultPartitionWriter[] getAllWriters() { return outputs.toArray(new ResultPartitionWriter[outputs.size()]); diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java index 1dac2e2d801..74d1e024a78 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java @@ -132,6 +132,7 @@ public class StreamTaskITCase { new ResultPartitionDeploymentDescriptor( PartitionDescriptorBuilder.newBuilder().build(), NettyShuffleDescriptorBuilder.newBuilder().buildLocal(), + 1, 1); return new TestTaskBuilder(shuffleEnvironment) .setInvokable(NoOpStreamTask.class)
