[FLINK-27609][Fix] Optimize REST call frequency

apache · May 16, 2022 · da03f8c · da03f8c
1 parent aa7bd6b
commit da03f8c
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 24 deletions.
diff --git a/.../org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java b/.../org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
@@ -88,25 +88,31 @@ public void observe(FlinkDeployment flinkApp, Context context) {
         }
 
         if (isJmDeploymentReady(flinkApp)) {
-            observeClusterInfo(flinkApp, observeConfig);
-            if (observeFlinkCluster(flinkApp, context, observeConfig)) {
-                if (reconciliationStatus.getState() != ReconciliationState.ROLLED_BACK) {
-                    reconciliationStatus.markReconciledSpecAsStable();
+            if (observeClusterInfo(flinkApp, observeConfig)) {
+                if (observeFlinkCluster(flinkApp, context, observeConfig)) {
+                    if (reconciliationStatus.getState() != ReconciliationState.ROLLED_BACK) {
+                        reconciliationStatus.markReconciledSpecAsStable();
+                    }
                 }
             }
         }
 
         clearErrorsIfDeploymentIsHealthy(flinkApp);
     }
 
-    private void observeClusterInfo(FlinkDeployment flinkApp, Configuration configuration) {
+    private boolean observeClusterInfo(FlinkDeployment flinkApp, Configuration configuration) {
+        if (flinkApp.getStatus().getClusterInfo() != null) {
+            return true;
+        }
         try {
-            Map<String, String> clusterInfo = flinkService.getClusterinfo(configuration);
+            Map<String, String> clusterInfo = flinkService.getClusterInfo(configuration);
             flinkApp.getStatus().setClusterInfo(clusterInfo);
             logger.debug("ClusterInfo: {}", clusterInfo);
         } catch (Exception e) {
-            logger.warn("Exception while fetching cluster info", e);
+            logger.error("Exception while fetching cluster info", e);
+            return false;
         }
+        return true;
     }
 
     protected void observeJmDeployment(
@@ -120,6 +126,8 @@ protected void observeJmDeployment(
             return;
         }
 
+        flinkApp.getStatus().setClusterInfo(null);
+
         logger.info(
                 "Observing JobManager deployment. Previous status: {}", previousJmStatus.name());
 

diff --git a/...tes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java b/...tes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
@@ -677,7 +677,7 @@ public SavepointFetchResult fetchSavepointInfo(
         }
     }
 
-    public Map<String, String> getClusterinfo(Configuration conf) throws Exception {
+    public Map<String, String> getClusterInfo(Configuration conf) throws Exception {
         Map<String, String> runtimeVersion = new HashMap<>();
 
         try (RestClusterClient<String> clusterClient =

diff --git a/...etes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java b/...etes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -355,7 +355,7 @@ public SubmittedJobInfo(
     }
 
     @Override
-    public Map<String, String> getClusterinfo(Configuration conf) throws Exception {
+    public Map<String, String> getClusterInfo(Configuration conf) throws Exception {
         return CLUSTER_INFO;
     }
 }
diff --git a/...ava/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java b/...ava/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
@@ -90,9 +90,11 @@ public void observeApplicationCluster() throws Exception {
                 JobManagerDeploymentStatus.DEPLOYED_NOT_READY,
                 deployment.getStatus().getJobManagerDeploymentStatus());
         assertNull(deployment.getStatus().getReconciliationStatus().getLastStableSpec());
+        assertNull(deployment.getStatus().getClusterInfo());
 
         // Stable ready
         observer.observe(deployment, readyContext);
+        assertEquals(TestingFlinkService.CLUSTER_INFO, deployment.getStatus().getClusterInfo());
         assertEquals(
                 JobManagerDeploymentStatus.READY,
                 deployment.getStatus().getJobManagerDeploymentStatus());
@@ -274,19 +276,4 @@ public void observeListJobsError() {
                         });
         assertEquals(podFailedMessage, exception.getMessage());
     }
-
-    @Test
-    public void observeClusterInfo() {
-        TestingFlinkService flinkService = new TestingFlinkService();
-        ApplicationObserver observer =
-                new ApplicationObserver(
-                        null, flinkService, configManager, new TestingStatusHelper<>());
-        FlinkDeployment deployment = TestUtils.buildApplicationCluster();
-        bringToReadyStatus(deployment);
-        observer.observe(deployment, readyContext);
-        assertEquals(
-                JobManagerDeploymentStatus.READY,
-                deployment.getStatus().getJobManagerDeploymentStatus());
-        assertEquals(TestingFlinkService.CLUSTER_INFO, deployment.getStatus().getClusterInfo());
-    }
 }