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 a643037 [FLINK-28478] Always complete in-progress update
a643037 is described below
commit a6430378553d7405a037f948576bea5ef5be55ce
Author: Gyula Fora <[email protected]>
AuthorDate: Thu Jul 14 14:00:33 2022 +0200
[FLINK-28478] Always complete in-progress update
---
.../AbstractFlinkResourceReconciler.java | 4 ++
.../kubernetes/operator/TestingFlinkService.java | 9 ++-
.../deployment/SessionReconcilerTest.java | 67 ++++++++++++++++++++--
3 files changed, 74 insertions(+), 6 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
index b0d8abe..7758ac6 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
@@ -270,6 +270,10 @@ public abstract class AbstractFlinkResourceReconciler<
* @return True if desired spec was already deployed.
*/
private boolean checkNewSpecAlreadyDeployed(CR resource, Configuration
deployConf) {
+ if (resource.getStatus().getReconciliationStatus().getState()
+ == ReconciliationState.UPGRADING) {
+ return false;
+ }
AbstractFlinkSpec deployedSpec =
ReconciliationUtils.getDeployedSpec(resource);
if (resource.getSpec().equals(deployedSpec)) {
LOG.info(
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
index 439b3ec..64a0b6d 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -144,6 +144,10 @@ public class TestingFlinkService extends FlinkService {
sessions.clear();
}
+ public Set<String> getSessions() {
+ return sessions;
+ }
+
@Override
public void submitApplicationCluster(
JobSpec jobSpec, Configuration conf, boolean requireHaMetadata)
throws Exception {
@@ -186,7 +190,10 @@ public class TestingFlinkService extends FlinkService {
}
@Override
- public void submitSessionCluster(Configuration conf) {
+ public void submitSessionCluster(Configuration conf) throws Exception {
+ if (deployFailure) {
+ throw new Exception("Deployment failure");
+ }
sessions.add(conf.get(KubernetesConfigOptions.CLUSTER_ID));
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java
index 2e13d28..dddd2eb 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java
@@ -23,16 +23,20 @@ import
org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.TestingStatusRecorder;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
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.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
/**
* Tests for {@link
org.apache.flink.kubernetes.operator.reconciler.deployment.SessionReconciler}.
@@ -41,17 +45,21 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
public class SessionReconcilerTest {
private final FlinkConfigManager configManager = new
FlinkConfigManager(new Configuration());
- private final EventRecorder eventRecorder = new EventRecorder(null, (r, e)
-> {});
private KubernetesClient kubernetesClient;
+ private EventRecorder eventRecorder;
+
+ @BeforeEach
+ public void before() {
+ eventRecorder = new EventRecorder(kubernetesClient, (r, e) -> {});
+ }
@Test
public void testStartSession() throws Exception {
- Context context = TestUtils.createEmptyContext();
var count = new AtomicInteger(0);
TestingFlinkService flinkService =
new TestingFlinkService() {
@Override
- public void submitSessionCluster(Configuration conf) {
+ public void submitSessionCluster(Configuration conf)
throws Exception {
super.submitSessionCluster(conf);
count.addAndGet(1);
}
@@ -66,7 +74,56 @@ public class SessionReconcilerTest {
new TestingStatusRecorder<>());
FlinkDeployment deployment = TestUtils.buildSessionCluster();
kubernetesClient.resource(deployment).createOrReplace();
- reconciler.reconcile(deployment, context);
+ reconciler.reconcile(deployment, flinkService.getContext());
assertEquals(1, count.get());
}
+
+ @Test
+ public void testFailedUpgrade() throws Exception {
+ var flinkService = new TestingFlinkService();
+ var reconciler =
+ new SessionReconciler(
+ kubernetesClient,
+ flinkService,
+ configManager,
+ eventRecorder,
+ new TestingStatusRecorder<>());
+
+ FlinkDeployment deployment = TestUtils.buildSessionCluster();
+ kubernetesClient.resource(deployment).createOrReplace();
+ reconciler.reconcile(deployment, flinkService.getContext());
+
+ assertEquals(
+ ReconciliationState.DEPLOYED,
+ deployment.getStatus().getReconciliationStatus().getState());
+
+ deployment.getSpec().setRestartNonce(1234L);
+
+ flinkService.setDeployFailure(true);
+ try {
+ reconciler.reconcile(deployment, flinkService.getContext());
+ fail();
+ } catch (Exception expected) {
+ }
+
+ assertEquals(
+ ReconciliationState.UPGRADING,
+ deployment.getStatus().getReconciliationStatus().getState());
+ flinkService.setDeployFailure(false);
+ flinkService.clear();
+ assertTrue(flinkService.getSessions().isEmpty());
+ reconciler.reconcile(deployment, flinkService.getContext());
+
+ assertEquals(
+ ReconciliationState.DEPLOYED,
+ deployment.getStatus().getReconciliationStatus().getState());
+ assertEquals(
+ 1234L,
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .deserializeLastReconciledSpec()
+ .getRestartNonce());
+ assertEquals(Set.of(deployment.getMetadata().getName()),
flinkService.getSessions());
+ }
}
