1996fanrui commented on code in PR #879:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/879#discussion_r1792893091
##########
flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobVertexScalerTest.java:
##########
@@ -775,11 +931,119 @@ public void
testSendingIneffectiveScalingEvents(Collection<ShipStrategy> inputSh
assertThat(event.getReason()).isEqualTo(INEFFECTIVE_SCALING);
}
+ @Test
+ public void testNumPartitionsAdjustment() {
+ final int minParallelism = 1;
+ final int maxParallelism = Integer.MAX_VALUE;
Review Comment:
They are `parallelismLowerLimit` and `parallelismUpperLimit`.
It's better to distinguish `parallelismUpperLimit` and `maxParallelism` here.
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java:
##########
@@ -383,23 +403,62 @@ protected static int scale(
// Apply min/max parallelism
newParallelism = Math.min(Math.max(parallelismLowerLimit,
newParallelism), upperBound);
- var adjustByMaxParallelism =
- inputShipStrategies.isEmpty() ||
inputShipStrategies.contains(HASH);
- if (!adjustByMaxParallelism) {
+ var adjustByMaxParallelismOrPartitions =
+ numSourcePartitions > 0 || inputShipStrategies.contains(HASH);
+ if (!adjustByMaxParallelismOrPartitions) {
return newParallelism;
}
- // When the shuffle type of vertex inputs contains keyBy or vertex is
a source, we try to
- // adjust the parallelism such that it divides the maxParallelism
without a remainder
- // => data is evenly spread across subtasks
- for (int p = newParallelism; p <= maxParallelism / 2 && p <=
upperBound; p++) {
- if (maxParallelism % p == 0) {
+ var numKeyGroupsOrPartitions =
+ numSourcePartitions <= 0 ? maxParallelism :
numSourcePartitions;
+ var upperBoundForAlignment =
+ Math.min(
+ // Optimize the case where newParallelism <=
maxParallelism / 2
+ newParallelism > numKeyGroupsOrPartitions / 2
+ ? numKeyGroupsOrPartitions
+ : numKeyGroupsOrPartitions / 2,
+ upperBound);
+
+ // When the shuffle type of vertex inputs contains keyBy or vertex is
a source,
+ // we try to adjust the parallelism such that it divides
+ // the numKeyGroupsOrPartitions without a remainder => data is evenly
spread across subtasks
+ for (int p = newParallelism; p <= upperBoundForAlignment; p++) {
+ if (numKeyGroupsOrPartitions % p == 0) {
return p;
}
}
- // If parallelism adjustment fails, use originally computed parallelism
- return newParallelism;
+ // When adjust the parallelism after rounding up cannot be evenly
divided by
+ // numKeyGroupsOrPartitions, Try to find the smallest parallelism that
can satisfy the
+ // current consumption rate.
+ int p = newParallelism;
+ for (; p > 0; p--) {
+ if (numKeyGroupsOrPartitions / p > numKeyGroupsOrPartitions /
newParallelism) {
+ if (numKeyGroupsOrPartitions % p != 0) {
+ p++;
+ }
+ break;
+ }
+ }
+
+ p = Math.max(p, parallelismLowerLimit);
+ var message =
+ String.format(
+ SCALE_LIMITED_MESSAGE_FORMAT,
+ vertex,
+ newParallelism,
+ p,
+ numSourcePartitions,
Review Comment:
After unifying, we should use `numKeyGroupsOrPartitions` instead of
`numSourcePartitions`, right?
If yes, the `SCALE_LIMITED_MESSAGE_FORMAT` needs to be updated together.
--
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]
