[FLINK-35278] Fix occasional NPE on getting latest resource for status replace

Author: ferenc-csaky

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.kubernetes.operator.utils;
 
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.lifecycle.ResourceLifecycleState;
@@ -127,47 +128,65 @@ private void replaceStatus(CR resource, STATUS prevStatus, KubernetesClient clie
             } catch (KubernetesClientException kce) {
                 // 409 is the error code for conflicts resulting from the locking
                 if (kce.getCode() == 409) {
-                    var currentVersion = resource.getMetadata().getResourceVersion();
-                    LOG.debug(
-                            "Could not apply status update for resource version {}",
-                            currentVersion);
-
-                    var latest = client.resource(resource).get();
-                    var latestVersion = latest.getMetadata().getResourceVersion();
-
-                    if (latestVersion.equals(currentVersion)) {
-                        // This should not happen as long as the client works consistently
-                        LOG.error("Unable to fetch latest resource version");
-                        throw kce;
-                    }
-
-                    if (latest.getStatus().equals(prevStatus)) {
-                        if (retries++ < 3) {
-                            LOG.debug(
-                                    "Retrying status update for latest version {}", latestVersion);
-                            resource.getMetadata().setResourceVersion(latestVersion);
-                        } else {
-                            // If we cannot get the latest version in 3 tries we throw the error to
-                            // retry with delay
-                            throw kce;
-                        }
-                    } else {
-                        throw new StatusConflictException(
-                                "Status have been modified externally in version "
-                                        + latestVersion
-                                        + " Previous: "
-                                        + objectMapper.writeValueAsString(prevStatus)
-                                        + " Latest: "
-                                        + objectMapper.writeValueAsString(latest.getStatus()));
-                    }
+                    handleLockingError(resource, prevStatus, client, retries, kce);
+                    ++retries;
                 } else {
-                    // We simply throw non conflict errors, to trigger retry with delay
+                    // We simply throw non-conflict errors, to trigger retry with delay
                     throw kce;
                 }
             }
         }
     }
 
+    @VisibleForTesting
+    void handleLockingError(
+            CR resource,
+            STATUS prevStatus,
+            KubernetesClient client,
+            int retries,
+            KubernetesClientException kce)
+            throws JsonProcessingException {
+
+        var currentVersion = resource.getMetadata().getResourceVersion();
+        LOG.debug("Could not apply status update for resource version {}", currentVersion);
+
+        var latest = client.resource(resource).get();
+        if (latest == null || latest.getMetadata() == null) {
+            // This can happen occasionally, we throw the error to retry with delay.
+            throw new KubernetesClientException(
+                    String.format(
+                            "Failed to retrieve latest %s",
+                            latest == null ? "resource" : "metadata"),
+                    kce);
+        }
+
+        var latestVersion = latest.getMetadata().getResourceVersion();
+        if (currentVersion.equals(latestVersion)) {
+            // This should not happen as long as the client works consistently
+            LOG.error("Unable to fetch latest resource version");
+            throw kce;
+        }
+
+        if (latest.getStatus().equals(prevStatus)) {
+            if (retries < 3) {
+                LOG.debug("Retrying status update for latest version {}", latestVersion);
+                resource.getMetadata().setResourceVersion(latestVersion);
+            } else {
+                // If we cannot get the latest version in 3 tries we throw the error to
+                // retry with delay
+                throw kce;
+            }
+        } else {
+            throw new StatusConflictException(
+                    "Status have been modified externally in version "
+                            + latestVersion
+                            + " Previous: "
+                            + objectMapper.writeValueAsString(prevStatus)
+                            + " Latest: "
+                            + objectMapper.writeValueAsString(latest.getStatus()));
+        }
+    }
+
     /**
      * Update the custom resource status based on the in-memory cached to ensure that any status
      * updates that we made previously are always visible in the reconciliation loop. This is

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java

@@ -24,11 +24,13 @@
 import org.apache.flink.kubernetes.operator.metrics.MetricManager;
 
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
 import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
 import org.junit.jupiter.api.Test;
 
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link StatusRecorder}. */
 @EnableKubernetesMockClient(crud = true)
@@ -47,17 +49,34 @@ public void testPatchOnlyWhenChanged() throws InterruptedException {
         var lastRequest = mockServer.getLastRequest();
 
         helper.patchAndCacheStatus(deployment, kubernetesClient);
-        assertTrue(mockServer.getLastRequest() != lastRequest);
+        assertThat(lastRequest).isNotSameAs(mockServer.getLastRequest());
         lastRequest = mockServer.getLastRequest();
         deployment.getStatus().getReconciliationStatus().setState(ReconciliationState.ROLLING_BACK);
         helper.patchAndCacheStatus(deployment, kubernetesClient);
 
         // We intentionally compare references
-        assertTrue(mockServer.getLastRequest() != lastRequest);
+        assertThat(lastRequest).isNotSameAs(mockServer.getLastRequest());
         lastRequest = mockServer.getLastRequest();
 
         // No update
         helper.patchAndCacheStatus(deployment, kubernetesClient);
-        assertTrue(mockServer.getLastRequest() == lastRequest);
+        assertThat(lastRequest).isSameAs(mockServer.getLastRequest());
+    }
+
+    @Test
+    public void testNullLatestResource() {
+        var statusRecorder =
+                new StatusRecorder<FlinkDeployment, FlinkDeploymentStatus>(
+                        new MetricManager<>(), (e, s) -> {});
+
+        var resource = TestUtils.buildApplicationCluster();
+        var cause = new KubernetesClientException("dummy");
+        assertThatThrownBy(
+                        () ->
+                                statusRecorder.handleLockingError(
+                                        resource, null, kubernetesClient, 0, cause))
+                .isInstanceOf(KubernetesClientException.class)
+                .hasMessage("Failed to retrieve latest resource")
+                .hasCause(cause);
     }
 }