This is an automated email from the ASF dual-hosted git repository. 1996fanrui pushed a commit to branch release-1.20 in repository https://gitbox.apache.org/repos/asf/flink.git
commit d302956a97ad42ef47e97471fea080f1e251ccbc 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. (cherry picked from commit 596c3732c7563f998e394473f12e24fc678e432e) --- .../generated/pipeline_configuration.html | 6 ++ .../flink/configuration/PipelineOptions.java | 62 +++++++++++++++++ .../flink/streaming/runtime/tasks/StreamTask.java | 66 +++++++++++++++--- .../tasks/StreamTaskMailboxTestHarnessBuilder.java | 7 ++ .../streaming/runtime/tasks/StreamTaskTest.java | 78 ++++++++++++++++++++++ 5 files changed, 208 insertions(+), 11 deletions(-) diff --git a/docs/layouts/shortcodes/generated/pipeline_configuration.html b/docs/layouts/shortcodes/generated/pipeline_configuration.html index f8b511206c7..f526bec9ea5 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 0c2a9f2aaa5..c987a5b1d2f 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 @@ -393,6 +393,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-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java index c3e51e72395..623bdab4214 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java @@ -23,6 +23,8 @@ import org.apache.flink.api.common.operators.MailboxExecutor; import org.apache.flink.api.common.operators.ProcessingTimeService.ProcessingTimeCallback; import org.apache.flink.configuration.CheckpointingOptions; import org.apache.flink.configuration.Configuration; +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.RestoreMode; import org.apache.flink.core.fs.AutoCloseableRegistry; @@ -1765,18 +1767,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/StreamTaskMailboxTestHarnessBuilder.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java index e16b2f73f49..987cb5ed08a 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java @@ -23,6 +23,7 @@ import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.checkpoint.TaskStateSnapshot; import org.apache.flink.runtime.execution.Environment; @@ -132,6 +133,12 @@ public class StreamTaskMailboxTestHarnessBuilder<OUT> { return this; } + public <T> StreamTaskMailboxTestHarnessBuilder<OUT> addJobConfig( + ConfigOption<T> option, T value) { + jobConfig.set(option, value); + return this; + } + public <T> StreamTaskMailboxTestHarnessBuilder<OUT> setCheckpointResponder( CheckpointResponder checkpointResponder) { this.checkpointResponder = checkpointResponder; 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 f1e5982d66b..2db95d19633 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 @@ -27,6 +27,8 @@ import org.apache.flink.api.common.state.CheckpointListener; import org.apache.flink.api.common.typeinfo.BasicTypeInfo; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.configuration.Configuration; +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); + } + } + private int getCurrentBufferSize(InputGate inputGate) { return getTestChannel(inputGate, 0).getCurrentBufferSize(); }
