You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/05/12 14:00:43 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-27573] Configuring a new random job result store directory
This is an automated email from the ASF dual-hosted git repository.
gyfora 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 73369b8 [FLINK-27573] Configuring a new random job result store directory
73369b8 is described below
commit 73369b851f2cd92a6818bb84e21157518d63a48d
Author: Matyas Orhidi <ma...@apple.com>
AuthorDate: Wed May 11 18:55:36 2022 +0200
[FLINK-27573] Configuring a new random job result store directory
---
docs/content/docs/concepts/overview.md | 15 ++++++++
examples/basic-checkpoint-ha.yaml | 1 +
.../operator/observer/JobStatusObserver.java | 11 ++++--
.../deployment/ApplicationReconciler.java | 29 +++++++++++++++-
.../deployment/ApplicationReconcilerTest.java | 40 ++++++++++++++++++++++
5 files changed, 92 insertions(+), 4 deletions(-)
diff --git a/docs/content/docs/concepts/overview.md b/docs/content/docs/concepts/overview.md
index 595d79c..e15f2e7 100644
--- a/docs/content/docs/concepts/overview.md
+++ b/docs/content/docs/concepts/overview.md
@@ -66,3 +66,18 @@ Flink Kubernetes Operator aims to capture the responsibilities of a human operat
| CI/CD | Continuous Integration | full | via github actions |
| | Public Docker repository | full | ghcr.io / dockerhub |
| | Public Helm repository | full | apache release repo |
+
+
+# Known issues & limitations
+
+JobResultStore Resource Leak
+
+To mitigate the impact of [FLINK-27569](https://issues.apache.org/jira/browse/FLINK-27569) the operator introduced a workaround [FLINK-27573](https://issues.apache.org/jira/browse/FLINK-27573) by setting `job-result-store.delete-on-commit=false` and a unique value for `job-result-store.storage-path` for every cluster launch. The storage path for older runs must be cleaned up manually, keeping the latest directory always:
+
+```shell
+ls -lth /tmp/flink/ha/job-result-store/basic-checkpoint-ha-example/
+total 0
+drwxr-xr-x 2 9999 9999 40 May 12 09:51 119e0203-c3a9-4121-9a60-d58839576f01 <- must be retained
+drwxr-xr-x 2 9999 9999 60 May 12 09:46 a6031ec7-ab3e-4b30-ba77-6498e58e6b7f
+drwxr-xr-x 2 9999 9999 60 May 11 15:11 b6fb2a9c-d1cd-4e65-a9a1-e825c4b47543
+```
diff --git a/examples/basic-checkpoint-ha.yaml b/examples/basic-checkpoint-ha.yaml
index 7454998..d9f32b4 100644
--- a/examples/basic-checkpoint-ha.yaml
+++ b/examples/basic-checkpoint-ha.yaml
@@ -28,6 +28,7 @@ spec:
state.savepoints.dir: file:///flink-data/savepoints
high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
high-availability.storageDir: file:///flink-data/ha
+ serviceAccount: flink
jobManager:
resource:
memory: "2048m"
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
index 6285c61..7eecb24 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
@@ -55,10 +55,10 @@ public abstract class JobStatusObserver<CTX> {
clusterJobStatuses = new ArrayList<>(flinkService.listJobs(deployedConfig));
} catch (Exception e) {
LOG.error("Exception while listing jobs", e);
- jobStatus.setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
if (e instanceof TimeoutException) {
onTimeout(ctx);
}
+ ifRunningMoveToReconciling(jobStatus, previousJobStatus);
return false;
}
@@ -79,12 +79,17 @@ public abstract class JobStatusObserver<CTX> {
}
return true;
} else {
- jobStatus.setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
- LOG.info("No job found on cluster yet");
+ ifRunningMoveToReconciling(jobStatus, previousJobStatus);
return false;
}
}
+ private void ifRunningMoveToReconciling(JobStatus jobStatus, String previousJobStatus) {
+ if (org.apache.flink.api.common.JobStatus.RUNNING.name().equals(previousJobStatus)) {
+ jobStatus.setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
+ }
+ }
+
/** Callback when list jobs timeout. */
protected abstract void onTimeout(CTX ctx);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
index c907ae1..8cd928f 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
@@ -17,7 +17,10 @@
package org.apache.flink.kubernetes.operator.reconciler.deployment;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
@@ -37,6 +40,7 @@ import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.operator.utils.IngressUtils;
import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
+import org.apache.flink.runtime.highavailability.JobResultStoreOptions;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
import io.fabric8.kubernetes.api.model.ObjectMeta;
@@ -46,6 +50,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Optional;
+import java.util.UUID;
/**
* Reconciler responsible for handling the job lifecycle according to the desired and current
@@ -208,7 +213,8 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
|| ReconciliationUtils.isJobRunning(status);
}
- private void deployFlinkJob(
+ @VisibleForTesting
+ protected void deployFlinkJob(
ObjectMeta meta,
JobSpec jobSpec,
FlinkDeploymentStatus status,
@@ -220,6 +226,9 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
} else {
effectiveConfig.removeConfig(SavepointConfigOptions.SAVEPOINT_PATH);
}
+
+ setRandomJobResultStorePath(effectiveConfig);
+
if (status.getJobManagerDeploymentStatus() != JobManagerDeploymentStatus.MISSING) {
if (!ReconciliationUtils.isJobInTerminalState(status)) {
throw new RuntimeException("This indicates a bug...");
@@ -253,6 +262,24 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
deployFlinkJob(meta, jobSpec, status, effectiveConfig, savepointOpt);
}
+ // Workaround for https://issues.apache.org/jira/browse/FLINK-27569
+ public static void setRandomJobResultStorePath(Configuration effectiveConfig) {
+ if (effectiveConfig.contains(HighAvailabilityOptions.HA_STORAGE_PATH)) {
+
+ if (!effectiveConfig.contains(JobResultStoreOptions.DELETE_ON_COMMIT)) {
+ effectiveConfig.set(JobResultStoreOptions.DELETE_ON_COMMIT, false);
+ }
+
+ effectiveConfig.set(
+ JobResultStoreOptions.STORAGE_PATH,
+ effectiveConfig.getString(HighAvailabilityOptions.HA_STORAGE_PATH)
+ + "/job-result-store/"
+ + effectiveConfig.getString(KubernetesConfigOptions.CLUSTER_ID)
+ + "/"
+ + UUID.randomUUID());
+ }
+ }
+
private void printCancelLogs(UpgradeMode upgradeMode) {
switch (upgradeMode) {
case STATELESS:
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
index b32e3fa..c4d0615 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
@@ -25,22 +25,29 @@ import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkVersion;
+import org.apache.flink.kubernetes.operator.crd.spec.JobSpec;
import org.apache.flink.kubernetes.operator.crd.spec.JobState;
import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
+import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.highavailability.JobResultStoreOptions;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.javaoperatorsdk.operator.api.reconciler.Context;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.ThreadLocalRandom;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -396,4 +403,37 @@ public class ApplicationReconcilerTest {
org.apache.flink.api.common.JobStatus.FINISHED.name(),
spDeployment.getStatus().getJobStatus().getState());
}
+
+ @Test
+ public void testRandomJobResultStorePath() throws Exception {
+
+ TestingFlinkService flinkService = new TestingFlinkService();
+ ApplicationReconciler reconciler =
+ new ApplicationReconciler(kubernetesClient, flinkService, configManager);
+ FlinkDeployment flinkApp = TestUtils.buildApplicationCluster();
+ final String haStoragePath = "file:///flink-data/ha";
+ flinkApp.getSpec()
+ .getFlinkConfiguration()
+ .put(HighAvailabilityOptions.HA_STORAGE_PATH.key(), haStoragePath);
+
+ ObjectMeta deployMeta = flinkApp.getMetadata();
+ FlinkDeploymentStatus status = flinkApp.getStatus();
+ FlinkDeploymentSpec spec = flinkApp.getSpec();
+ JobSpec jobSpec = spec.getJob();
+ Configuration deployConfig = configManager.getDeployConfig(deployMeta, spec);
+
+ status.getJobStatus().setState(org.apache.flink.api.common.JobStatus.FINISHED.name());
+ status.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+ reconciler.deployFlinkJob(deployMeta, jobSpec, status, deployConfig, Optional.empty());
+
+ String path1 = deployConfig.get(JobResultStoreOptions.STORAGE_PATH);
+ Assertions.assertTrue(path1.startsWith(haStoragePath));
+
+ status.getJobStatus().setState(org.apache.flink.api.common.JobStatus.FINISHED.name());
+ status.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+ reconciler.deployFlinkJob(deployMeta, jobSpec, status, deployConfig, Optional.empty());
+ String path2 = deployConfig.get(JobResultStoreOptions.STORAGE_PATH);
+ Assertions.assertTrue(path2.startsWith(haStoragePath));
+ assertNotEquals(path1, path2);
+ }
}