1996fanrui commented on code in PR #29272:
URL: https://github.com/apache/flink/pull/29272#discussion_r4287434265
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -305,6 +305,68 @@ public enum VertexDescriptionMode {
CASCADING
}
+ public static final ConfigOption<ForwardEdgeParallelismMismatchMode>
+ FORWARD_EDGE_PARALLELISM_MISMATCH_MODE =
+ key("pipeline.forward-edge.parallelism-mismatch-mode")
+ .enumType(ForwardEdgeParallelismMismatchMode.class)
+
.defaultValue(ForwardEdgeParallelismMismatchMode.REBALANCE)
Review Comment:
Good point. REBALANCE stays the default to keep the existing behavior, so
jobs relying on it today (e.g. append-only jobs rescaled by the
AdaptiveScheduler) aren't affected. Changelog jobs can opt into FAIL or
KEEP_FORWARD.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -2046,13 +2049,59 @@
List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters
private static void
replaceForwardPartitionerIfConsumerParallelismDoesNotMatch(
Environment environment, NonChainedOutput streamOutput, int
outputIndex) {
- if (streamOutput.getPartitioner() instanceof ForwardPartitioner
- &&
environment.getWriter(outputIndex).getNumberOfSubpartitions()
- !=
environment.getTaskInfo().getNumberOfParallelSubtasks()) {
- LOG.debug(
- "Replacing forward partitioner with rebalance for {}",
- environment.getTaskInfo().getTaskNameWithSubtasks());
- streamOutput.setPartitioner(new RebalancePartitioner<>());
+ 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)
+ // An undecided consumer parallelism is not a mismatch.
+ || consumerParallelism
+ ==
ResultPartitionDeploymentDescriptor.UNKNOWN_CONSUMER_PARALLELISM
+ || consumerParallelism == producerParallelism) {
Review Comment:
I'd prefer each mode to mean exactly one behavior:
- REBALANCE always rebalances
- KEEP_FORWARD always keeps forward
Special-casing downscale would make REBALANCE silently behave like
`KEEP_FORWARD` and may bring other risks we haven't thought of. Users who want
that can set KEEP_FORWARD.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -2046,13 +2049,59 @@
List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters
private static void
replaceForwardPartitionerIfConsumerParallelismDoesNotMatch(
Environment environment, NonChainedOutput streamOutput, int
outputIndex) {
- if (streamOutput.getPartitioner() instanceof ForwardPartitioner
- &&
environment.getWriter(outputIndex).getNumberOfSubpartitions()
- !=
environment.getTaskInfo().getNumberOfParallelSubtasks()) {
- LOG.debug(
- "Replacing forward partitioner with rebalance for {}",
- environment.getTaskInfo().getTaskNameWithSubtasks());
- streamOutput.setPartitioner(new RebalancePartitioner<>());
+ 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)
+ // An undecided consumer parallelism is not a mismatch.
+ || 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.debug(
+ "Replacing forward partitioner with rebalance for {} "
+ + "(producer parallelism {} != consumer
parallelism {}).",
+ taskNameWithSubtasks,
+ producerParallelism,
+ consumerParallelism);
+ 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:
Review Comment:
FAIL is for setups where a parallelism mismatch is never expected, so it's
surfaced immediately instead of being handled silently.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -2046,13 +2049,59 @@
List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters
private static void
replaceForwardPartitionerIfConsumerParallelismDoesNotMatch(
Environment environment, NonChainedOutput streamOutput, int
outputIndex) {
- if (streamOutput.getPartitioner() instanceof ForwardPartitioner
- &&
environment.getWriter(outputIndex).getNumberOfSubpartitions()
- !=
environment.getTaskInfo().getNumberOfParallelSubtasks()) {
- LOG.debug(
- "Replacing forward partitioner with rebalance for {}",
- environment.getTaskInfo().getTaskNameWithSubtasks());
- streamOutput.setPartitioner(new RebalancePartitioner<>());
+ 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)
+ // An undecided consumer parallelism is not a mismatch.
+ || 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.debug(
Review Comment:
Agreed, changed it to WARN and the message now states that records may be
reordered and how to avoid it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
