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 96f55807499550d89ddfb90d928811cd8cc09370 Author: Rui Fan <[email protected]> AuthorDate: Fri Oct 2 23:44:13 2026 +0200 [FLINK-40904][runtime] Detect FORWARD parallelism mismatch with the actual consumer parallelism The FORWARD-to-REBALANCE downgrade compared the producer parallelism with the number of subpartitions, which is not the consumer parallelism on a POINTWISE edge. Equal parallelism > 1 was taken as a mismatch, and N -> M with N > 1 could go undetected. Compare with the actual consumer parallelism instead and include both values in the log. --- .../flink/streaming/runtime/tasks/StreamTask.java | 18 ++++++++++++++---- .../streaming/runtime/tasks/StreamMockEnvironment.java | 14 +++++++++++++- .../flink/streaming/runtime/tasks/StreamTaskTest.java | 14 ++++++++++---- 3 files changed, 37 insertions(+), 9 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java index b7f12973948..8a08c73a887 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java @@ -49,6 +49,7 @@ import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo; import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger; import org.apache.flink.runtime.checkpoint.channel.SequentialChannelStateReader; import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager; +import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor; import org.apache.flink.runtime.execution.CancelTaskException; import org.apache.flink.runtime.execution.Environment; import org.apache.flink.runtime.io.AvailabilityProvider; @@ -2046,12 +2047,21 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> private static void replaceForwardPartitionerIfConsumerParallelismDoesNotMatch( Environment environment, NonChainedOutput streamOutput, int outputIndex) { + final int producerParallelism = environment.getTaskInfo().getNumberOfParallelSubtasks(); + // The number of subpartitions isn't the consumer parallelism on a POINTWISE edge, so the + // actual consumer parallelism is passed through the deployment descriptor. + final int consumerParallelism = environment.getWriterConsumerParallelism(outputIndex); if (streamOutput.getPartitioner() instanceof ForwardPartitioner - && environment.getWriter(outputIndex).getNumberOfSubpartitions() - != environment.getTaskInfo().getNumberOfParallelSubtasks()) { + // An undecided consumer parallelism is not a mismatch. + && consumerParallelism + != ResultPartitionDeploymentDescriptor.UNKNOWN_CONSUMER_PARALLELISM + && consumerParallelism != producerParallelism) { LOG.debug( - "Replacing forward partitioner with rebalance for {}", - environment.getTaskInfo().getTaskNameWithSubtasks()); + "Replacing forward partitioner with rebalance for {} " + + "(producer parallelism {} != consumer parallelism {}).", + environment.getTaskInfo().getTaskNameWithSubtasks(), + producerParallelism, + consumerParallelism); streamOutput.setPartitioner(new RebalancePartitioner<>()); } } 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 27f37f24cf7..06bce51026a 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 @@ -90,7 +90,7 @@ public class StreamMockEnvironment implements Environment { private final JobInfo jobInfo; - private final TaskInfo taskInfo; + private TaskInfo taskInfo; private final MemoryManager memManager; @@ -337,6 +337,18 @@ public class StreamMockEnvironment implements Environment { this.writerConsumerParallelism = writerConsumerParallelism; } + public void setNumberOfParallelSubtasks(int numberOfParallelSubtasks) { + this.taskInfo = + new TaskInfoImpl( + taskInfo.getTaskName(), + Math.max( + taskInfo.getMaxNumberOfParallelSubtasks(), + numberOfParallelSubtasks), + taskInfo.getIndexOfThisSubtask(), + numberOfParallelSubtasks, + taskInfo.getAttemptNumber()); + } + @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/StreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java index 1b42449f539..93d000b8ec5 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java @@ -154,6 +154,8 @@ import org.apache.flink.util.function.SupplierWithException; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import org.mockito.ArgumentCaptor; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; @@ -1838,8 +1840,10 @@ public class StreamTaskTest { } } - @Test - void testForwardPartitionerIsConvertedToRebalanceOnParallelismChanges() throws Exception { + @ParameterizedTest(name = "producer parallelism {0}, consumer parallelism {1}") + @CsvSource({"1, 2", "2, 1"}) + void testForwardPartitionerIsConvertedToRebalanceOnParallelismChanges( + int producerParallelism, int consumerParallelism) throws Exception { StreamTaskMailboxTestHarnessBuilder<Integer> builder = new StreamTaskMailboxTestHarnessBuilder<>( OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO) @@ -1861,16 +1865,18 @@ public class StreamTaskTest { .getChannelSelector()) .isInstanceOf(ForwardPartitioner.class); - // Simulate changed downstream task parallelism (1->2) + // Simulate changed producer and consumer parallelism List<ResultPartitionWriter> newOutputs = new ArrayList<>(); newOutputs.add( new MockResultPartitionWriter() { @Override public int getNumberOfSubpartitions() { - return 2; + return consumerParallelism; } }); harness.streamMockEnvironment.setOutputs(newOutputs); + harness.streamMockEnvironment.setNumberOfParallelSubtasks(producerParallelism); + harness.streamMockEnvironment.setWriterConsumerParallelism(consumerParallelism); // Re-create outputs recordWriterDelegate =
