You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2024/05/09 08:00:23 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-34212][autoscaler] Autoscaler Standalone cleans up stopped jobs to prevent memory leaks (#824)
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 04829bf1 [FLINK-34212][autoscaler] Autoscaler Standalone cleans up stopped jobs to prevent memory leaks (#824)
04829bf1 is described below
commit 04829bf1fadb45d8f76849f4263596c01b94fee3
Author: Rui Fan <19...@gmail.com>
AuthorDate: Thu May 9 16:00:17 2024 +0800
[FLINK-34212][autoscaler] Autoscaler Standalone cleans up stopped jobs to prevent memory leaks (#824)
---
.../standalone/StandaloneAutoscalerExecutor.java | 71 ++++++--
.../StandaloneAutoscalerExecutorTest.java | 180 +++++++++++++++++++++
2 files changed, 236 insertions(+), 15 deletions(-)
diff --git a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java
index 741be348..a075b718 100644
--- a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java
+++ b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java
@@ -35,14 +35,18 @@ import javax.annotation.Nonnull;
import java.io.Closeable;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.CONTROL_LOOP_INTERVAL;
import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.CONTROL_LOOP_PARALLELISM;
@@ -64,11 +68,17 @@ public class StandaloneAutoscalerExecutor<KEY, Context extends JobAutoScalerCont
private final UnmodifiableConfiguration baseConf;
/**
- * Maintain a set of job keys that during scaling, it should be updated at {@link
+ * Maintain a set of job keys that during scaling, it should be accessed at {@link
* #scheduledExecutorService} thread.
*/
private final Set<KEY> scalingJobKeys;
+ /**
+ * Maintain a set of scaling job keys for the last control loop, it should be accessed at {@link
+ * #scheduledExecutorService} thread.
+ */
+ private Set<KEY> lastScalingKeys;
+
public StandaloneAutoscalerExecutor(
@Nonnull Configuration conf,
@Nonnull JobListFetcher<KEY, Context> jobListFetcher,
@@ -105,36 +115,67 @@ public class StandaloneAutoscalerExecutor<KEY, Context extends JobAutoScalerCont
scalingThreadPool.shutdownNow();
}
+ /**
+ * @return All CompletableFuture for all scaling jobs, note: it's only used for test for now.
+ */
@VisibleForTesting
- protected void scaling() {
+ protected List<CompletableFuture<Void>> scaling() {
LOG.info("Standalone autoscaler starts scaling.");
Collection<Context> jobList;
try {
jobList = jobListFetcher.fetch(baseConf);
} catch (Throwable e) {
LOG.error("Error while fetch job list.", e);
- return;
+ return Collections.emptyList();
}
+ cleanupStoppedJob(jobList);
+
+ var resultFutures = new ArrayList<CompletableFuture<Void>>();
for (var jobContext : jobList) {
final var jobKey = jobContext.getJobKey();
if (scalingJobKeys.contains(jobKey)) {
continue;
}
scalingJobKeys.add(jobKey);
- CompletableFuture.runAsync(() -> scalingSingleJob(jobContext), scalingThreadPool)
- .whenCompleteAsync(
- (result, throwable) -> {
- if (throwable != null) {
- LOG.error(
- "Error while jobKey: {} executing scaling .",
- jobKey,
- throwable);
- }
- scalingJobKeys.remove(jobKey);
- },
- scheduledExecutorService);
+ resultFutures.add(
+ CompletableFuture.runAsync(
+ () -> scalingSingleJob(jobContext), scalingThreadPool)
+ .whenCompleteAsync(
+ (result, throwable) -> {
+ if (throwable != null) {
+ LOG.error(
+ "Error while jobKey: {} executing scaling .",
+ jobKey,
+ throwable);
+ }
+ scalingJobKeys.remove(jobKey);
+ if (!lastScalingKeys.contains(jobKey)) {
+ // Current job has been stopped. lastScalingKeys doesn't
+ // contain jobKey means current job key was scaled in a
+ // previous control loop, and current job is stopped in
+ // the latest control loop.
+ autoScaler.cleanup(jobKey);
+ }
+ },
+ scheduledExecutorService));
+ }
+ return resultFutures;
+ }
+
+ private void cleanupStoppedJob(Collection<Context> jobList) {
+ var currentScalingKeys =
+ jobList.stream().map(JobAutoScalerContext::getJobKey).collect(Collectors.toSet());
+ if (lastScalingKeys != null) {
+ lastScalingKeys.removeAll(currentScalingKeys);
+ for (KEY jobKey : lastScalingKeys) {
+ // Current job may be scaling, and cleanup should happen after scaling.
+ if (!scalingJobKeys.contains(jobKey)) {
+ autoScaler.cleanup(jobKey);
+ }
+ }
}
+ lastScalingKeys = currentScalingKeys;
}
@VisibleForTesting
diff --git a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutorTest.java b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutorTest.java
index 74154a72..caffdbbb 100644
--- a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutorTest.java
+++ b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutorTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.autoscaler.JobAutoScaler;
import org.apache.flink.autoscaler.JobAutoScalerContext;
import org.apache.flink.autoscaler.event.TestingEventCollector;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.concurrent.FutureUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -33,8 +34,10 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
@@ -227,6 +230,183 @@ class StandaloneAutoscalerExecutorTest {
}
}
+ @Test
+ void testCleanupAfterStopped() throws Exception {
+ var eventCollector = new TestingEventCollector<JobID, JobAutoScalerContext<JobID>>();
+
+ var job1 = createJobAutoScalerContext();
+ var job2 = createJobAutoScalerContext();
+ var scaleCounter = new ConcurrentHashMap<JobID, Integer>();
+ var cleanupCounter = new ConcurrentHashMap<JobID, Integer>();
+
+ var jobList = new ArrayList<JobAutoScalerContext<JobID>>();
+
+ try (var autoscalerExecutor =
+ new StandaloneAutoscalerExecutor<>(
+ new Configuration(),
+ baseConf -> jobList,
+ eventCollector,
+ new JobAutoScaler<>() {
+ @Override
+ public void scale(JobAutoScalerContext<JobID> context) {
+ scaleCounter.put(
+ context.getJobKey(),
+ scaleCounter.getOrDefault(context.getJobKey(), 0) + 1);
+ }
+
+ @Override
+ public void cleanup(JobID jobID) {
+ cleanupCounter.put(
+ jobID, cleanupCounter.getOrDefault(jobID, 0) + 1);
+ }
+ })) {
+
+ // Test for empty job list.
+ var scaledFutures = autoscalerExecutor.scaling();
+ assertThat(scaledFutures).isEmpty();
+ FutureUtils.waitForAll(scaledFutures).get();
+ assertThat(scaleCounter).isEmpty();
+ assertThat(cleanupCounter).isEmpty();
+
+ // Test for 2 jobs twice.
+ jobList.add(job1);
+ jobList.add(job2);
+
+ scaledFutures = autoscalerExecutor.scaling();
+ assertThat(scaledFutures).hasSize(2);
+ FutureUtils.waitForAll(scaledFutures).get();
+ assertThat(scaleCounter)
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(job1.getJobKey(), 1, job2.getJobKey(), 1));
+ assertThat(cleanupCounter).isEmpty();
+
+ scaledFutures = autoscalerExecutor.scaling();
+ assertThat(scaledFutures).hasSize(2);
+ FutureUtils.waitForAll(scaledFutures).get();
+ assertThat(scaleCounter)
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(job1.getJobKey(), 2, job2.getJobKey(), 2));
+ assertThat(cleanupCounter).isEmpty();
+
+ // Test for 1 job twice.
+ jobList.clear();
+ jobList.add(job2);
+
+ scaledFutures = autoscalerExecutor.scaling();
+ assertThat(scaledFutures).hasSize(1);
+ FutureUtils.waitForAll(scaledFutures).get();
+ assertThat(scaleCounter)
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(job1.getJobKey(), 2, job2.getJobKey(), 3));
+ assertThat(cleanupCounter)
+ .containsExactlyInAnyOrderEntriesOf(Map.of(job1.getJobKey(), 1));
+
+ scaledFutures = autoscalerExecutor.scaling();
+ assertThat(scaledFutures).hasSize(1);
+ FutureUtils.waitForAll(scaledFutures).get();
+ assertThat(scaleCounter)
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(job1.getJobKey(), 2, job2.getJobKey(), 4));
+ // Only cleanup once.
+ assertThat(cleanupCounter)
+ .containsExactlyInAnyOrderEntriesOf(Map.of(job1.getJobKey(), 1));
+ }
+ }
+
+ @Test
+ void testCleanupForStoppedJobAfterScaling() throws Exception {
+ var eventCollector = new TestingEventCollector<JobID, JobAutoScalerContext<JobID>>();
+
+ var job1 = createJobAutoScalerContext();
+ var job2 = createJobAutoScalerContext();
+ var scaleCounter = new ConcurrentHashMap<JobID, Integer>();
+ var cleanupCounter = new ConcurrentHashMap<JobID, Integer>();
+
+ var job1StartWaitFuture = new CompletableFuture<>();
+ var job1WaitFuture = new CompletableFuture<>();
+
+ var jobList = new ArrayList<JobAutoScalerContext<JobID>>();
+
+ try (var autoscalerExecutor =
+ new StandaloneAutoscalerExecutor<>(
+ new Configuration(),
+ baseConf -> jobList,
+ eventCollector,
+ new JobAutoScaler<>() {
+ @Override
+ public void scale(JobAutoScalerContext<JobID> context)
+ throws Exception {
+ scaleCounter.put(
+ context.getJobKey(),
+ scaleCounter.getOrDefault(context.getJobKey(), 0) + 1);
+ if (context == job1) {
+ job1StartWaitFuture.complete(null);
+ job1WaitFuture.get();
+ }
+ }
+
+ @Override
+ public void cleanup(JobID jobID) {
+ cleanupCounter.put(
+ jobID, cleanupCounter.getOrDefault(jobID, 0) + 1);
+ }
+ })) {
+
+ // Test for job1 and job2
+ jobList.add(job1);
+ jobList.add(job2);
+
+ var scaledFutures = autoscalerExecutor.scaling();
+ var job1ScaledFuture = scaledFutures.get(0);
+
+ assertThat(scaledFutures).hasSize(2);
+ // wait for job2 scaling is finished.
+ scaledFutures.get(1).get();
+ // wait for job1 starts wait.
+ job1StartWaitFuture.get();
+ assertThat(scaleCounter)
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(job1.getJobKey(), 1, job2.getJobKey(), 1));
+ assertThat(cleanupCounter).isEmpty();
+
+ scalingOnlyHappenForJob2(
+ job1, job2, scaleCounter, cleanupCounter, autoscalerExecutor, 2);
+
+ // Test for job2 twice, and job1 should be clean up after scaling.
+ jobList.clear();
+ jobList.add(job2);
+
+ scalingOnlyHappenForJob2(
+ job1, job2, scaleCounter, cleanupCounter, autoscalerExecutor, 3);
+ scalingOnlyHappenForJob2(
+ job1, job2, scaleCounter, cleanupCounter, autoscalerExecutor, 4);
+
+ // Wait for job1 scaling to complete.
+ job1WaitFuture.complete(null);
+ job1ScaledFuture.get();
+ assertThat(cleanupCounter)
+ .containsExactlyInAnyOrderEntriesOf(Map.of(job1.getJobKey(), 1));
+ }
+ }
+
+ private void scalingOnlyHappenForJob2(
+ JobAutoScalerContext<JobID> job1,
+ JobAutoScalerContext<JobID> job2,
+ ConcurrentHashMap<JobID, Integer> scaleCounter,
+ ConcurrentHashMap<JobID, Integer> cleanupCounter,
+ StandaloneAutoscalerExecutor<JobID, JobAutoScalerContext<JobID>> autoscalerExecutor,
+ int expectedJob2ScaleCounter)
+ throws Exception {
+ var scaledFutures = autoscalerExecutor.scaling();
+ assertThat(scaledFutures).hasSize(1);
+ FutureUtils.waitForAll(scaledFutures).get();
+ assertThat(scaleCounter)
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(job1.getJobKey(), 1, job2.getJobKey(), expectedJob2ScaleCounter));
+ // job1 is still scaling, it cannot be cleaned up.
+ assertThat(cleanupCounter).isEmpty();
+ }
+
private static JobAutoScalerContext<JobID> createJobAutoScalerContext() {
var jobID = new JobID();
return new JobAutoScalerContext<>(