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 2023/09/13 12:09:48 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-33011] Never accidentally delete HA metadata for last state deployments
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 82739f62 [FLINK-33011] Never accidentally delete HA metadata for last state deployments
82739f62 is described below
commit 82739f62adda33e686da7d8aa30cbd41ea13012f
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Wed Sep 13 14:09:42 2023 +0200
[FLINK-33011] Never accidentally delete HA metadata for last state deployments
---
.../deployment/ApplicationReconciler.java | 2 +-
.../deployment/ApplicationReconcilerTest.java | 42 ++++++++++++++++++++++
2 files changed, 43 insertions(+), 1 deletion(-)
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 5ba2154a..cc5594c2 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
@@ -165,7 +165,7 @@ public class ApplicationReconciler
Preconditions.checkArgument(ReconciliationUtils.isJobInTerminalState(status));
LOG.info("Deleting deployment with terminated application before new deployment");
flinkService.deleteClusterDeployment(
- relatedResource.getMetadata(), status, deployConfig, true);
+ relatedResource.getMetadata(), status, deployConfig, !requireHaMetadata);
flinkService.waitForClusterShutdown(deployConfig);
statusRecorder.patchAndCacheStatus(relatedResource, ctx.getKubernetesClient());
}
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 1f03ee70..a0bea65d 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
@@ -32,6 +32,7 @@ import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.operator.OperatorTestBase;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
import org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec;
@@ -71,6 +72,7 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.JobVertexResourceRequirements;
import org.apache.flink.util.concurrent.Executors;
+import io.fabric8.kubernetes.api.model.DeletionPropagation;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
@@ -80,6 +82,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.junit.platform.commons.util.StringUtils;
import java.time.Clock;
@@ -92,6 +95,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
@@ -999,6 +1003,44 @@ public class ApplicationReconcilerTest extends OperatorTestBase {
assertEquals(JobManagerDeploymentStatus.MISSING, status.getJobManagerDeploymentStatus());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testClusterCleanupBeforeDeploy(boolean requireMetadata) throws Exception {
+ var flinkApp = TestUtils.buildApplicationCluster();
+ var status = flinkApp.getStatus();
+ var spec = flinkApp.getSpec();
+ var deployConfig = configManager.getDeployConfig(flinkApp.getMetadata(), spec);
+
+ status.getReconciliationStatus().serializeAndSetLastReconciledSpec(spec, flinkApp);
+ status.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+ status.getJobStatus().setState(org.apache.flink.api.common.JobStatus.FINISHED.name());
+
+ var deleted = new AtomicBoolean(false);
+
+ flinkService =
+ new TestingFlinkService() {
+ @Override
+ protected void deleteClusterInternal(
+ ObjectMeta meta,
+ Configuration conf,
+ boolean deleteHaMeta,
+ DeletionPropagation deletionPropagation) {
+ deleted.set(deleteHaMeta);
+ }
+ };
+
+ reconciler
+ .getReconciler()
+ .deploy(
+ getResourceContext(flinkApp),
+ spec,
+ deployConfig,
+ Optional.empty(),
+ requireMetadata);
+ assertEquals(deleted.get(), !requireMetadata);
+ assertEquals(JobManagerDeploymentStatus.DEPLOYING, status.getJobManagerDeploymentStatus());
+ }
+
@Test
public void testDeploymentRecoveryEvent() throws Exception {
FlinkDeployment deployment = TestUtils.buildApplicationCluster();