mxm commented on code in PR #879:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/879#discussion_r1757006171
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java:
##########
@@ -389,15 +413,51 @@ protected static int scale(
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) {
+ int adjustableMaxParallelism = maxParallelism;
+ int adjustableUpperBound;
+ if (numPartitions <= 0) {
+ adjustableUpperBound = Math.min(maxParallelism / 2, upperBound);
+ } else {
+ adjustableUpperBound = Math.min(numPartitions, upperBound);
+ adjustableMaxParallelism = numPartitions;
+ }
+
+ // 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 adjustableMaxParallelism without a remainder => data is evenly
spread across subtasks
+ for (int p = newParallelism; p <= adjustableUpperBound; p++) {
+ if (adjustableMaxParallelism % p == 0) {
return p;
}
}
+ if (numPartitions > 0) {
+
+ // When adjust the parallelism after rounding up cannot be evenly
divided by source
+ // numPartitions, Try to find the smallest parallelism that can
satisfy the current
+ // consumption rate.
+ int finalParallelism = newParallelism;
+ for (; finalParallelism > parallelismLowerLimit;
finalParallelism--) {
+ if (numPartitions / finalParallelism > numPartitions /
newParallelism) {
+ if (numPartitions % finalParallelism != 0) {
+ finalParallelism++;
+ }
+ break;
+ }
+ }
+ consumer.accept(
+ String.format(
+ SCALE_LIMITED_MESSAGE_FORMAT,
+ vertex,
+ newParallelism,
+ finalParallelism,
+ String.format(
+ "numPartitions :
%s，upperBound(maxParallelism or parallelismUpperLimit): %s, "
+ + "parallelismLowerLimit: %s.",
+ numPartitions, upperBound,
parallelismLowerLimit)));
Review Comment:
There is a test implementation of the event handler which we can simply pass
in for tests.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
