Efrat19 commented on code in PR #29272:
URL: https://github.com/apache/flink/pull/29272#discussion_r4280200554
##########
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:
Nit? as it can be harmful for changelog
```suggestion
LOG.warn(
```
##########
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:
IIUC `REBALANCE` breaks changelog mode - should it be the default?
##########
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:
wdyt about
```suggestion
|| consumerParallelism <= producerParallelism) {
```
maybe we can keep the forward partitioner in this case as well, since each
producer will still forward to 1 consumer?
##########
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:
I don't understand why do we need the fail mode?
Wouldn't it always be preferred to downgrade performance with warn
(`KEEP_FORWARD`) other than crash?
--
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]
