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 596c3732c7563f998e394473f12e24fc678e432e Author: Rui Fan <[email protected]> AuthorDate: Wed Oct 21 10:33:20 2026 +0200 [FLINK-40780][runtime] Make FORWARD-to-REBALANCE downgrade on parallelism mismatch configurable When a FORWARD edge connects a producer and consumer whose parallelism no longer match (e.g. after a rescale via the AdaptiveScheduler or pipeline.jobvertex-parallelism-overrides), the runtime silently replaces the ForwardPartitioner with a RebalancePartitioner (FLINK-30213). This is safe for append-only streams but silently reorders order-sensitive (changelog) streams, producing incorrect results. Add pipeline.forward-edge.parallelism-mismatch-mode to control this behavior: - REBALANCE (default): current behavior, warn log. - FAIL: reject the job with an exception. - KEEP_FORWARD: keep the forward partitioner (order-preserving), warn log. --- .../generated/pipeline_configuration.html | 6 ++ .../flink/configuration/PipelineOptions.java | 62 +++++++++++++++++ .../flink/streaming/runtime/tasks/StreamTask.java | 66 +++++++++++++++--- .../streaming/runtime/tasks/StreamTaskTest.java | 78 ++++++++++++++++++++++ 4 files changed, 201 insertions(+), 11 deletions(-) diff --git a/docs/layouts/shortcodes/generated/pipeline_configuration.html b/docs/layouts/shortcodes/generated/pipeline_configuration.html index e4050aee746..4338f8fe827 100644 --- a/docs/layouts/shortcodes/generated/pipeline_configuration.html +++ b/docs/layouts/shortcodes/generated/pipeline_configuration.html @@ -56,6 +56,12 @@ <td>Boolean</td> <td>Forces Flink to register avro classes in kryo serializer.<br /><br />Important: Make sure to include the flink-avro module. Otherwise, nothing will be registered. For backward compatibility, the default value is empty to conform to the behavior of the older version. That is, always register avro with kryo, and if flink-avro is not in the class path, register a dummy serializer. In Flink-2.0, we will set the default value to true.</td> </tr> + <tr> + <td><h5>pipeline.forward-edge.parallelism-mismatch-mode</h5></td> + <td style="word-wrap: break-word;">REBALANCE</td> + <td><p>Enum</p></td> + <td>Determines how the runtime handles a FORWARD (pointwise) edge whose producer and consumer parallelism no longer match, which can happen when the parallelism of connected operators is changed independently (for example via the AdaptiveScheduler or <code class="highlighter-rouge">pipeline.jobvertex-parallelism-overrides</code>). A FORWARD edge is only valid at equal parallelism; on a mismatch one of the following strategies is applied:<ul><li><code class="highlighter-rouge" [...] + </tr> <tr> <td><h5>pipeline.generic-types</h5></td> <td style="word-wrap: break-word;">true</td> diff --git a/flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java index 4742b6dbca8..d366b63ce0e 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java @@ -305,6 +305,68 @@ public class PipelineOptions { CASCADING } + public static final ConfigOption<ForwardEdgeParallelismMismatchMode> + FORWARD_EDGE_PARALLELISM_MISMATCH_MODE = + key("pipeline.forward-edge.parallelism-mismatch-mode") + .enumType(ForwardEdgeParallelismMismatchMode.class) + .defaultValue(ForwardEdgeParallelismMismatchMode.REBALANCE) + .withDescription( + Description.builder() + .text( + "Determines how the runtime handles a FORWARD (pointwise) edge whose" + + " producer and consumer parallelism no longer match, which can happen" + + " when the parallelism of connected operators is changed" + + " independently (for example via the AdaptiveScheduler or %s). A" + + " FORWARD edge is only valid at equal parallelism; on a mismatch one" + + " of the following strategies is applied:", + code(PARALLELISM_OVERRIDES.key())) + .list( + text( + "%s: replace the forward partitioner with a rebalance (round-robin)" + + " partitioner. This preserves throughput but reorders records," + + " which is safe for append-only streams but corrupts" + + " order-sensitive (changelog) streams.", + code( + ForwardEdgeParallelismMismatchMode + .REBALANCE + .name())), + text( + "%s: fail the job with an exception instead of silently changing" + + " the data distribution.", + code( + ForwardEdgeParallelismMismatchMode + .FAIL + .name())), + text( + "%s: keep the forward partitioner. Record order is preserved but" + + " the records are funneled to a single consumer subtask, so the" + + " remaining consumer subtasks stay idle.", + code( + ForwardEdgeParallelismMismatchMode + .KEEP_FORWARD + .name()))) + .build()); + + /** + * The strategy applied when a FORWARD (pointwise) edge connects a producer and consumer with + * mismatched parallelism. + */ + @PublicEvolving + public enum ForwardEdgeParallelismMismatchMode { + /** + * Replace the forward partitioner with a rebalance partitioner. Reorders records; safe only + * for append-only streams. + */ + REBALANCE, + /** Fail the job with an exception. */ + FAIL, + /** + * Keep the forward partitioner. Preserves record order but funnels records to a single + * consumer subtask. + */ + KEEP_FORWARD + } + public static final ConfigOption<Boolean> VERTEX_NAME_INCLUDE_INDEX_PREFIX = key("pipeline.vertex-name-include-index-prefix") .booleanType() 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 8a08c73a887..df3a58bdd71 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 @@ -25,6 +25,8 @@ import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.configuration.CheckpointingOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.NettyShuffleEnvironmentOptions; +import org.apache.flink.configuration.PipelineOptions; +import org.apache.flink.configuration.PipelineOptions.ForwardEdgeParallelismMismatchMode; import org.apache.flink.configuration.TaskManagerOptions; import org.apache.flink.core.execution.RecoveryClaimMode; import org.apache.flink.core.fs.AutoCloseableRegistry; @@ -2051,18 +2053,60 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> // 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 + if (!(streamOutput.getPartitioner() instanceof ForwardPartitioner) // An undecided consumer parallelism is not a mismatch. - && consumerParallelism - != ResultPartitionDeploymentDescriptor.UNKNOWN_CONSUMER_PARALLELISM - && consumerParallelism != producerParallelism) { - LOG.debug( - "Replacing forward partitioner with rebalance for {} " - + "(producer parallelism {} != consumer parallelism {}).", - environment.getTaskInfo().getTaskNameWithSubtasks(), - producerParallelism, - consumerParallelism); - streamOutput.setPartitioner(new RebalancePartitioner<>()); + || consumerParallelism + == ResultPartitionDeploymentDescriptor.UNKNOWN_CONSUMER_PARALLELISM + || consumerParallelism == producerParallelism) { + return; + } + + final ForwardEdgeParallelismMismatchMode mode = + environment + .getJobConfiguration() + .get(PipelineOptions.FORWARD_EDGE_PARALLELISM_MISMATCH_MODE); + final String taskNameWithSubtasks = environment.getTaskInfo().getTaskNameWithSubtasks(); + switch (mode) { + case REBALANCE: + LOG.warn( + "Replacing forward partitioner with rebalance for {} " + + "(producer parallelism {} != consumer parallelism {}). This may " + + "reorder records and produce incorrect results for order-sensitive " + + "(changelog) streams. Set '{}' to {} or {} to avoid it.", + taskNameWithSubtasks, + producerParallelism, + consumerParallelism, + PipelineOptions.FORWARD_EDGE_PARALLELISM_MISMATCH_MODE.key(), + ForwardEdgeParallelismMismatchMode.FAIL, + ForwardEdgeParallelismMismatchMode.KEEP_FORWARD); + streamOutput.setPartitioner(new RebalancePartitioner<>()); + break; + case KEEP_FORWARD: + LOG.warn( + "Keeping forward partitioner for {} despite a parallelism mismatch " + + "(producer parallelism {} != consumer parallelism {}). Record order " + + "is preserved but records are funneled to a single consumer subtask, " + + "leaving the remaining consumer subtasks idle.", + taskNameWithSubtasks, + producerParallelism, + consumerParallelism); + break; + case FAIL: + throw new FlinkRuntimeException( + String.format( + "Forward partitioning cannot be preserved across a parallelism change " + + "for %s (producer parallelism %d != consumer parallelism %d). " + + "Silently downgrading a FORWARD edge to a redistributing " + + "exchange can reorder records and corrupt order-sensitive " + + "(changelog) results. Set %s to %s or %s to allow it.", + taskNameWithSubtasks, + producerParallelism, + consumerParallelism, + PipelineOptions.FORWARD_EDGE_PARALLELISM_MISMATCH_MODE.key(), + ForwardEdgeParallelismMismatchMode.REBALANCE, + ForwardEdgeParallelismMismatchMode.KEEP_FORWARD)); + default: + throw new IllegalStateException("Unhandled mode: " + mode); } } 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 93d000b8ec5..8ca889c033a 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 @@ -30,6 +30,8 @@ import org.apache.flink.configuration.CheckpointingOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.IllegalConfigurationException; import org.apache.flink.configuration.NettyShuffleEnvironmentOptions; +import org.apache.flink.configuration.PipelineOptions; +import org.apache.flink.configuration.PipelineOptions.ForwardEdgeParallelismMismatchMode; import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.core.execution.SavepointFormatType; import org.apache.flink.core.fs.FSDataInputStream; @@ -1893,6 +1895,82 @@ public class StreamTaskTest { } } + @Test + void testForwardPartitionerFailsOnParallelismChangeWhenModeIsFail() throws Exception { + StreamTaskMailboxTestHarnessBuilder<Integer> builder = + new StreamTaskMailboxTestHarnessBuilder<>( + OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO) + .addInput(BasicTypeInfo.INT_TYPE_INFO) + .addJobConfig( + PipelineOptions.FORWARD_EDGE_PARALLELISM_MISMATCH_MODE, + ForwardEdgeParallelismMismatchMode.FAIL) + .setOutputPartitioner(new ForwardPartitioner<>()) + .setupOutputForSingletonOperatorChain( + new TestBoundedOneInputStreamOperator()); + + try (StreamTaskMailboxTestHarness<Integer> harness = builder.build()) { + // Simulate changed downstream task parallelism (1->2) + List<ResultPartitionWriter> newOutputs = new ArrayList<>(); + newOutputs.add( + new MockResultPartitionWriter() { + @Override + public int getNumberOfSubpartitions() { + return 2; + } + }); + harness.streamMockEnvironment.setOutputs(newOutputs); + harness.streamMockEnvironment.setWriterConsumerParallelism(2); + + // With FAIL mode the job is rejected instead of silently downgrading the edge. + assertThatThrownBy( + () -> + harness.streamTask.createRecordWriterDelegate( + harness.streamTask.configuration, + harness.streamMockEnvironment)) + .isInstanceOf(FlinkRuntimeException.class) + .hasMessageContaining("Forward partitioning cannot be preserved"); + } + } + + @Test + void testForwardPartitionerIsKeptOnParallelismChangeWhenModeIsKeepForward() throws Exception { + StreamTaskMailboxTestHarnessBuilder<Integer> builder = + new StreamTaskMailboxTestHarnessBuilder<>( + OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO) + .addInput(BasicTypeInfo.INT_TYPE_INFO) + .addJobConfig( + PipelineOptions.FORWARD_EDGE_PARALLELISM_MISMATCH_MODE, + ForwardEdgeParallelismMismatchMode.KEEP_FORWARD) + .setOutputPartitioner(new ForwardPartitioner<>()) + .setupOutputForSingletonOperatorChain( + new TestBoundedOneInputStreamOperator()); + + try (StreamTaskMailboxTestHarness<Integer> harness = builder.build()) { + // Simulate changed downstream task parallelism (1->2) + List<ResultPartitionWriter> newOutputs = new ArrayList<>(); + newOutputs.add( + new MockResultPartitionWriter() { + @Override + public int getNumberOfSubpartitions() { + return 2; + } + }); + harness.streamMockEnvironment.setOutputs(newOutputs); + harness.streamMockEnvironment.setWriterConsumerParallelism(2); + + RecordWriterDelegate<SerializationDelegate<StreamRecord<Object>>> recordWriterDelegate = + harness.streamTask.createRecordWriterDelegate( + harness.streamTask.configuration, harness.streamMockEnvironment); + // With KEEP_FORWARD mode the forward partitioner is preserved despite the mismatch. + assertThat( + ((ChannelSelectorRecordWriter) + ((SingleRecordWriter) recordWriterDelegate) + .getRecordWriter(0)) + .getChannelSelector()) + .isInstanceOf(ForwardPartitioner.class); + } + } + @Test void testGetAndCheckMaxTraverseSize() { Configuration config = new Configuration();
