You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "1996fanrui (via GitHub)" <gi...@apache.org> on 2024/12/09 03:25:16 UTC
[PR] [FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down [flink-kubernetes-operator]
1996fanrui opened a new pull request, #922:
URL: https://github.com/apache/flink-kubernetes-operator/pull/922
## What is the purpose of the change
[FLINK-36863][autoscaler] Use the maximum parallelism in the **_past scale-down.interval window_** when scaling down
## Brief change log
VertexDelayedScaleDownInfo maintain all recommended parallelisms at each time within the past scale-down.interval window period.
- Evicts the recommended parallelism before the scale-down.interval window.
- The max parallelism within the window range as the final parallelism.
It is a scenario that calculates the max value within a sliding window.
- It is similar with [leetcode 239: Sliding Window Maximum](https://leetcode.com/problems/sliding-window-maximum/description/).
- If latest parallelism is greater than the past parallelism, the past parallelism never be the max value, so we could evict all smaller parallelism in the past.
- We only need to maintain a list with monotonically decreasing parallelism within the past window.
- The first parallelism is the final parallelism.
## Verifying this change
Doing
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changes to the `CustomResourceDescriptors`: no
- Core observer or reconciler logic that is regularly executed: no
## Documentation
- Does this pull request introduce a new feature? no
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #922:
URL: https://github.com/apache/flink-kubernetes-operator/pull/922#issuecomment-2586680918
> @1996fanrui it may make sense to merge #921 first and then adjust and merge this one. As it may need a rebase as you also refactored some utilization related bits
Thanks @gyfora for the review and reminder! I will merge #921 tomorrow if no more comments~
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #922:
URL: https://github.com/apache/flink-kubernetes-operator/pull/922#discussion_r1901559620
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java:
##########
@@ -259,6 +220,9 @@ Map<JobVertexID, ScalingSummary> computeScalingSummary(
if (parallelismChange.isNoChange()) {
return;
}
+ if (parallelismChange.isOutsideUtilizationBound()) {
+ anyVertexOutsideBound.set(true);
+ }
Review Comment:
We check the `OutsideUtilizationBound` of the final recommended parallelism instead of the `OutsideUtilizationBound` of latest recommended parallelism.
If the Utilization of all tasks is within range, we can skip scaling.
##########
flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownEndToEndTest.java:
##########
@@ -0,0 +1,723 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.autoscaler;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.autoscaler.config.AutoScalerOptions;
+import org.apache.flink.autoscaler.event.TestingEventCollector;
+import org.apache.flink.autoscaler.metrics.ScalingMetric;
+import org.apache.flink.autoscaler.metrics.TestMetrics;
+import org.apache.flink.autoscaler.realizer.TestingScalingRealizer;
+import org.apache.flink.autoscaler.state.AutoScalerStateStore;
+import org.apache.flink.autoscaler.state.InMemoryAutoScalerStateStore;
+import org.apache.flink.autoscaler.topology.JobTopology;
+import org.apache.flink.autoscaler.topology.VertexInfo;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext;
+import static org.apache.flink.autoscaler.TestingAutoscalerUtils.getRestClusterClientSupplier;
+import static org.apache.flink.autoscaler.metrics.ScalingMetric.RECOMMENDED_PARALLELISM;
+import static org.apache.flink.autoscaler.topology.ShipStrategy.REBALANCE;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end test for {@link DelayedScaleDown}. */
+public class DelayedScaleDownEndToEndTest {
+
+ private static final int INITIAL_SOURCE_PARALLELISM = 200;
+ private static final int INITIAL_SINK_PARALLELISM = 1000;
+ private static final double UTILIZATION_TARGET = 0.8;
+
+ private JobAutoScalerContext<JobID> context;
+ private AutoScalerStateStore<JobID, JobAutoScalerContext<JobID>> stateStore;
+
+ TestingScalingRealizer<JobID, JobAutoScalerContext<JobID>> scalingRealizer;
+
+ private TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>> metricsCollector;
+
+ private JobVertexID source, sink;
+
+ private JobAutoScalerImpl<JobID, JobAutoScalerContext<JobID>> autoscaler;
+ private Instant now;
+ private int expectedMetricSize;
+
+ @BeforeEach
+ public void setup() throws Exception {
+ context = createDefaultJobAutoScalerContext();
+
+ TestingEventCollector<JobID, JobAutoScalerContext<JobID>> eventCollector =
+ new TestingEventCollector<>();
+ stateStore = new InMemoryAutoScalerStateStore<>();
+
+ source = new JobVertexID();
+ sink = new JobVertexID();
+
+ metricsCollector =
+ new TestingMetricsCollector<>(
+ new JobTopology(
+ new VertexInfo(source, Map.of(), INITIAL_SOURCE_PARALLELISM, 4000),
+ new VertexInfo(
+ sink,
+ Map.of(source, REBALANCE),
+ INITIAL_SINK_PARALLELISM,
+ 4000)));
+
+ var scaleDownInterval = Duration.ofMinutes(60).minus(Duration.ofSeconds(1));
+ // The metric window size is 9:59 to avoid other metrics are mixed.
+ var metricWindow = Duration.ofMinutes(10).minus(Duration.ofSeconds(1));
+
+ var defaultConf = context.getConfiguration();
+ defaultConf.set(AutoScalerOptions.AUTOSCALER_ENABLED, true);
+ defaultConf.set(AutoScalerOptions.SCALING_ENABLED, true);
+ defaultConf.set(AutoScalerOptions.STABILIZATION_INTERVAL, Duration.ZERO);
+ defaultConf.set(AutoScalerOptions.RESTART_TIME, Duration.ofSeconds(0));
+ defaultConf.set(AutoScalerOptions.CATCH_UP_DURATION, Duration.ofSeconds(0));
+ defaultConf.set(AutoScalerOptions.VERTEX_MAX_PARALLELISM, 10000);
+ defaultConf.set(AutoScalerOptions.SCALING_ENABLED, true);
+ defaultConf.set(AutoScalerOptions.MAX_SCALE_DOWN_FACTOR, 1.);
+ defaultConf.set(AutoScalerOptions.MAX_SCALE_UP_FACTOR, (double) Integer.MAX_VALUE);
+ defaultConf.set(AutoScalerOptions.TARGET_UTILIZATION, UTILIZATION_TARGET);
+ defaultConf.set(AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY, 0.1);
+ defaultConf.set(AutoScalerOptions.SCALE_DOWN_INTERVAL, scaleDownInterval);
+ defaultConf.set(AutoScalerOptions.METRICS_WINDOW, metricWindow);
+
+ scalingRealizer = new TestingScalingRealizer<>();
+ autoscaler =
+ new JobAutoScalerImpl<>(
+ metricsCollector,
+ new ScalingMetricEvaluator(),
+ new ScalingExecutor<>(eventCollector, stateStore),
+ eventCollector,
+ scalingRealizer,
+ stateStore);
+
+ // initially the last evaluated metrics are empty
+ assertThat(autoscaler.lastEvaluatedMetrics.get(context.getJobKey())).isNull();
+
+ now = Instant.ofEpochMilli(0);
+ setClocksTo(now);
+ running(now);
+
+ metricsCollector.updateMetrics(source, buildMetric(0, 800));
+ metricsCollector.updateMetrics(sink, buildMetric(0, 800));
+
+ // the recommended parallelism values are empty initially
+ autoscaler.scale(context);
+ expectedMetricSize = 1;
+ assertCollectedMetricsSize(expectedMetricSize);
+ }
+
+ /**
+ * The scale down won't be executed before scale down interval window is full, and it will use
+ * the max recommended parallelism in the past scale down interval window size when scale down
+ * is executed.
+ */
+ @Test
+ void testDelayedScaleDownHappenInLastMetricWindow() throws Exception {
+ var sourceBusyList = List.of(800, 800, 800, 800, 800, 800, 800);
+ var sinkBusyList = List.of(350, 300, 150, 200, 400, 250, 100);
+
+ var metricWindowSize = sourceBusyList.size();
+
+ assertThat(metricWindowSize).isGreaterThan(6);
+
+ var totalRecords = 0L;
+ int recordsPerMinutes = 4800000;
+
+ for (int windowIndex = 0; windowIndex < metricWindowSize; windowIndex++) {
+ for (int i = 1; i <= 10; i++) {
+ now = now.plus(Duration.ofMinutes(1));
+ setClocksTo(now);
+
+ metricsCollector.updateMetrics(
+ source, buildMetric(totalRecords, sourceBusyList.get(windowIndex)));
+ metricsCollector.updateMetrics(
+ sink, buildMetric(totalRecords, sinkBusyList.get(windowIndex)));
+
+ autoscaler.scale(context);
+ // Metric window is 10 minutes, so 10 is the maximal metric size.
+ expectedMetricSize = Math.min(expectedMetricSize + 1, 10);
+ assertCollectedMetricsSize(expectedMetricSize);
+
+ // Assert the recommended parallelism.
+ if (windowIndex == metricWindowSize - 1 && i == 10) {
+ // Last metric, we expect scale down is executed, and max recommended
+ // parallelism in the past window should be used.
+ // The max busy time needs more parallelism than others, so we could compute
+ // parallelism based on the max busy time.
+ var expectedSourceParallelism =
+ getExpectedParallelism(sourceBusyList, INITIAL_SOURCE_PARALLELISM);
+ var expectedSinkParallelism =
+ getExpectedParallelism(sinkBusyList, INITIAL_SINK_PARALLELISM);
+ pollAndAssertScalingRealizer(
+ expectedSourceParallelism, expectedSinkParallelism);
+ } else {
+ // Otherwise, scale down cannot be executed.
+ if (windowIndex == 0 && i <= 9) {
+ // Metric window is not full, so don't have recommended parallelism.
+ assertThat(getCurrentMetricValue(source, RECOMMENDED_PARALLELISM)).isNull();
+ assertThat(getCurrentMetricValue(sink, RECOMMENDED_PARALLELISM)).isNull();
+ } else {
+ // Scale down won't be executed before scale down interval window is full.
+ assertThat(getCurrentMetricValue(source, RECOMMENDED_PARALLELISM))
+ .isEqualTo(INITIAL_SOURCE_PARALLELISM);
+ assertThat(getCurrentMetricValue(sink, RECOMMENDED_PARALLELISM))
+ .isEqualTo(INITIAL_SINK_PARALLELISM);
+ }
+ assertThat(scalingRealizer.events).isEmpty();
+ }
+
+ totalRecords += recordsPerMinutes;
+ }
+ }
+ }
+
+ private static List<List<Integer>>
+ sinkParallelismMaxRecommendedParallelismWithinUtilizationBoundaryProvider() {
+ return List.of(
+ List.of(
+ 700, 690, 690, 690, 690, 690, 700, 690, 690, 690, 690, 690, 700, 690, 690,
+ 690, 690, 690, 700, 690, 690, 690, 690, 690, 700, 690, 690, 690, 690, 690,
+ 700),
+ List.of(700, 690, 700, 690, 700, 690, 700, 690, 700, 690, 700, 690, 700, 690, 700),
+ List.of(
+ 790, 200, 200, 200, 200, 200, 790, 200, 200, 200, 200, 200, 790, 200, 200,
+ 200, 200, 200, 790, 200, 200, 200, 200, 200, 790, 200, 200, 200, 200, 200,
+ 790),
+ List.of(790, 200, 790, 200, 790, 200, 790, 200, 790, 200, 790, 200, 790, 200, 790));
Review Comment:
If the `RecommendedParallelism` class does not include `outsideUtilizationBound`, this test will be failed for the last 2 inputs.
The scale down interval is 60 minutes, and metric window is 10 minutes, so the recommended parallelism will be 790 for the last 2 inputs. When we check `outsideUtilizationBound`, we need to check the `outsideUtilizationBound` corresponding to the parallelism of 790.
That's why `RecommendedParallelism` class includes `outsideUtilizationBound`. If it doesn't include `outsideUtilizationBound`, the latest `outsideUtilizationBound` may come from parallelism 200
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #922:
URL: https://github.com/apache/flink-kubernetes-operator/pull/922#issuecomment-2589438165
Thanks for the review, merging~
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui merged PR #922:
URL: https://github.com/apache/flink-kubernetes-operator/pull/922
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #922:
URL: https://github.com/apache/flink-kubernetes-operator/pull/922#discussion_r1881460715
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/DelayedScaleDown.java:
##########
@@ -30,23 +30,78 @@
import java.time.Instant;
import java.util.HashMap;
+import java.util.LinkedList;
import java.util.Map;
+import static org.apache.flink.util.Preconditions.checkState;
+
/** All delayed scale down requests. */
public class DelayedScaleDown {
+ @Data
+ private static class RecommendedParallelism {
+ @Nonnull private final Instant triggerTime;
+ private final int parallelism;
+
+ @JsonCreator
+ public RecommendedParallelism(
+ @Nonnull @JsonProperty("triggerTime") Instant triggerTime,
+ @JsonProperty("parallelism") int parallelism) {
+ this.triggerTime = triggerTime;
+ this.parallelism = parallelism;
+ }
+ }
+
/** The delayed scale down info for vertex. */
@Data
public static class VertexDelayedScaleDownInfo {
private final Instant firstTriggerTime;
- private int maxRecommendedParallelism;
+ // TODO : add the comment to explain how to calculate the max parallelism within the sliding
+ // window.
+ private final LinkedList<RecommendedParallelism> recommendedParallelisms;
+
+ public VertexDelayedScaleDownInfo(Instant firstTriggerTime) {
+ this.firstTriggerTime = firstTriggerTime;
+ this.recommendedParallelisms = new LinkedList<>();
+ }
@JsonCreator
public VertexDelayedScaleDownInfo(
@JsonProperty("firstTriggerTime") Instant firstTriggerTime,
- @JsonProperty("maxRecommendedParallelism") int maxRecommendedParallelism) {
+ @JsonProperty("recommendedParallelisms")
+ LinkedList<RecommendedParallelism> recommendedParallelisms) {
this.firstTriggerTime = firstTriggerTime;
- this.maxRecommendedParallelism = maxRecommendedParallelism;
+ this.recommendedParallelisms = recommendedParallelisms;
+ }
+
+ /** Record current recommended parallelism. */
+ public void recordRecommendedParallelism(Instant triggerTime, int parallelism) {
+
+ // Remove all recommended parallelisms that are lower than the latest parallelism.
+ while (!recommendedParallelisms.isEmpty()
+ && recommendedParallelisms.peekLast().getParallelism() <= parallelism) {
+ recommendedParallelisms.pollLast();
+ }
Review Comment:
This is a good optimization 👍
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #922:
URL: https://github.com/apache/flink-kubernetes-operator/pull/922#discussion_r1881470489
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/DelayedScaleDown.java:
##########
@@ -30,23 +30,78 @@
import java.time.Instant;
import java.util.HashMap;
+import java.util.LinkedList;
import java.util.Map;
+import static org.apache.flink.util.Preconditions.checkState;
+
/** All delayed scale down requests. */
public class DelayedScaleDown {
+ @Data
+ private static class RecommendedParallelism {
+ @Nonnull private final Instant triggerTime;
+ private final int parallelism;
+
+ @JsonCreator
+ public RecommendedParallelism(
+ @Nonnull @JsonProperty("triggerTime") Instant triggerTime,
+ @JsonProperty("parallelism") int parallelism) {
+ this.triggerTime = triggerTime;
+ this.parallelism = parallelism;
+ }
+ }
+
/** The delayed scale down info for vertex. */
@Data
public static class VertexDelayedScaleDownInfo {
private final Instant firstTriggerTime;
- private int maxRecommendedParallelism;
+ // TODO : add the comment to explain how to calculate the max parallelism within the sliding
+ // window.
+ private final LinkedList<RecommendedParallelism> recommendedParallelisms;
+
+ public VertexDelayedScaleDownInfo(Instant firstTriggerTime) {
+ this.firstTriggerTime = firstTriggerTime;
+ this.recommendedParallelisms = new LinkedList<>();
+ }
@JsonCreator
public VertexDelayedScaleDownInfo(
@JsonProperty("firstTriggerTime") Instant firstTriggerTime,
- @JsonProperty("maxRecommendedParallelism") int maxRecommendedParallelism) {
+ @JsonProperty("recommendedParallelisms")
+ LinkedList<RecommendedParallelism> recommendedParallelisms) {
this.firstTriggerTime = firstTriggerTime;
- this.maxRecommendedParallelism = maxRecommendedParallelism;
+ this.recommendedParallelisms = recommendedParallelisms;
+ }
+
+ /** Record current recommended parallelism. */
+ public void recordRecommendedParallelism(Instant triggerTime, int parallelism) {
+
+ // Remove all recommended parallelisms that are lower than the latest parallelism.
+ while (!recommendedParallelisms.isEmpty()
+ && recommendedParallelisms.peekLast().getParallelism() <= parallelism) {
+ recommendedParallelisms.pollLast();
+ }
Review Comment:
Thanks @gyfora for the comment in advance, I'm still adding more tests and comments for this PR.
I have some other works to do this week, so I expect this PR will be ready next week.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #922:
URL: https://github.com/apache/flink-kubernetes-operator/pull/922#discussion_r1881470489
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/DelayedScaleDown.java:
##########
@@ -30,23 +30,78 @@
import java.time.Instant;
import java.util.HashMap;
+import java.util.LinkedList;
import java.util.Map;
+import static org.apache.flink.util.Preconditions.checkState;
+
/** All delayed scale down requests. */
public class DelayedScaleDown {
+ @Data
+ private static class RecommendedParallelism {
+ @Nonnull private final Instant triggerTime;
+ private final int parallelism;
+
+ @JsonCreator
+ public RecommendedParallelism(
+ @Nonnull @JsonProperty("triggerTime") Instant triggerTime,
+ @JsonProperty("parallelism") int parallelism) {
+ this.triggerTime = triggerTime;
+ this.parallelism = parallelism;
+ }
+ }
+
/** The delayed scale down info for vertex. */
@Data
public static class VertexDelayedScaleDownInfo {
private final Instant firstTriggerTime;
- private int maxRecommendedParallelism;
+ // TODO : add the comment to explain how to calculate the max parallelism within the sliding
+ // window.
+ private final LinkedList<RecommendedParallelism> recommendedParallelisms;
+
+ public VertexDelayedScaleDownInfo(Instant firstTriggerTime) {
+ this.firstTriggerTime = firstTriggerTime;
+ this.recommendedParallelisms = new LinkedList<>();
+ }
@JsonCreator
public VertexDelayedScaleDownInfo(
@JsonProperty("firstTriggerTime") Instant firstTriggerTime,
- @JsonProperty("maxRecommendedParallelism") int maxRecommendedParallelism) {
+ @JsonProperty("recommendedParallelisms")
+ LinkedList<RecommendedParallelism> recommendedParallelisms) {
this.firstTriggerTime = firstTriggerTime;
- this.maxRecommendedParallelism = maxRecommendedParallelism;
+ this.recommendedParallelisms = recommendedParallelisms;
+ }
+
+ /** Record current recommended parallelism. */
+ public void recordRecommendedParallelism(Instant triggerTime, int parallelism) {
+
+ // Remove all recommended parallelisms that are lower than the latest parallelism.
+ while (!recommendedParallelisms.isEmpty()
+ && recommendedParallelisms.peekLast().getParallelism() <= parallelism) {
+ recommendedParallelisms.pollLast();
+ }
Review Comment:
Thanks @gyfora for the comment in advanced, I'm still adding more tests and comments for this PR.
I have some other works to do this week, so I expect this PR will be ready next week.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on PR #922:
URL: https://github.com/apache/flink-kubernetes-operator/pull/922#issuecomment-2586427779
@1996fanrui it may make sense to merge https://github.com/apache/flink-kubernetes-operator/pull/921 first and then adjust and merge this one. As it may need a rebase as you also refactored some utilization related bits
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org