[SPARK-56228] Fix StatusRecorder to refresh resourceVersion on 409 Error before retrying

### What changes were proposed in this pull request?

This PR fixes the `StatusRecorder.patchAndStatusWithVersionLocked()` retry loop to refresh the `resourceVersion` from the API server when a 409 Conflict occurs, before retrying the status update.

### Why are the changes needed?

When a 409 Conflict (`"the object has been modified; please apply your changes to the latest version"`) occurs during a status update, the retry loop retries with the same stale `resourceVersion`, causing all subsequent retries to fail with the same conflict error. The fix fetches the latest resource from the API server on 409 to obtain the current `resourceVersion`, following the existing pattern used in `ReconcilerUtils`.

```
26/03/26 02:10:34 ERROR pi default o.a.s.k.o.u.StatusRecorder
Error while persisting status to
ApplicationStatus(super=BaseStatus(currentState=ApplicationState(super=BaseState(currentStateSummary=ResourceReleased, lastTransitionTime=2026-03-26T02:10:31.642213107Z, message=null), lastObservedDriverStatus=null), stateTransitionHistory={0=ApplicationState(super=BaseState(currentStatio.fabric8.kubernetes.client.KubernetesClientException:
Failure executing: PUT at: https://10.43.0.1:443/apis/spark.apache.org/v1/namespaces/default/sparkapplications/pi/status.
Message: Operation cannot be fulfilled on sparkapplications.spark.apache.org "pi":
the object has been modified; please apply your changes to the latest version and try again. Received status: Status(apiV
at io.fabric8.kubernetes.client.KubernetesClientException.copyAsCause(KubernetesClientException.java:205)
at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.waitForResult(OperationSupport.java:507)
at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.handleResponse(OperationSupport.java:524)
at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.handleUpdate(OperationSupport.java:358)
at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.handleUpdate(BaseOperation.java:771)
at io.fabric8.kubernetes.client.dsl.internal.HasMetadataOperation.update(HasMetadataOperation.java:138)
at io.fabric8.kubernetes.client.dsl.internal.HasMetadataOperation.update(HasMetadataOperation.java:121)
at io.fabric8.kubernetes.client.dsl.internal.HasMetadataOperation.updateStatus(HasMetadataOperation.java:126)
at io.fabric8.kubernetes.client.dsl.internal.HasMetadataOperation.updateStatus(HasMetadataOperation.java:44)
at org.apache.spark.k8s.operator.utils.StatusRecorder.patchAndStatusWithVersionLocked(StatusRecorder.java:99)
```

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

A new unit test `refreshesResourceVersionOn409Conflict` was added to `StatusRecorderTest` to verify that the `resourceVersion` is refreshed after a 409 Conflict and the retry succeeds.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (claude-opus-4-6)

Closes #592 from dongjoon-hyun/SPARK-56228.

Authored-by: Dongjoon Hyun <dongjoon@apache.org>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: dongjoon-hyun

spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/StatusRecorder.java

@@ -19,6 +19,7 @@
 
 package org.apache.spark.k8s.operator.utils;
 
+import static java.net.HttpURLConnection.HTTP_CONFLICT;
 import static org.apache.spark.k8s.operator.config.SparkOperatorConf.API_RETRY_ATTEMPT_AFTER_SECONDS;
 import static org.apache.spark.k8s.operator.config.SparkOperatorConf.API_STATUS_PATCH_MAX_ATTEMPTS;
 
@@ -102,6 +103,18 @@ private void patchAndStatusWithVersionLocked(CR resource, KubernetesClient clien
         break;
       } catch (KubernetesClientException e) {
         log.debug("Error while patching status, retrying {}/{}...", i, maxRetry, e);
+        if (e.getCode() == HTTP_CONFLICT) {
+          try {
+            CR latest = client.resource(resource).get();
+            if (latest != null) {
+              resource
+                  .getMetadata()
+                  .setResourceVersion(latest.getMetadata().getResourceVersion());
+            }
+          } catch (KubernetesClientException refreshEx) {
+            log.debug("Failed to refresh resource version", refreshEx);
+          }
+        }
         try {
           Thread.sleep(TimeUnit.SECONDS.toMillis(API_RETRY_ATTEMPT_AFTER_SECONDS.getValue()));
         } catch (InterruptedException ie) {

spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/StatusRecorderTest.java

@@ -56,6 +56,37 @@ class StatusRecorderTest {
       new StatusRecorder<>(
           List.of(mockStatusListener), ApplicationStatus.class, SparkApplication.class);
 
+  @Test
+  void refreshesResourceVersionOn409Conflict() {
+    var testResource = getSparkApplication("1");
+    var resourceV2 = getSparkApplication("2");
+    var resourceV3 = getSparkApplication("3");
+
+    BaseContext<SparkApplication> context = mock(BaseContext.class);
+    when(context.getResource()).thenReturn(testResource);
+    when(context.getClient()).thenReturn(client);
+    var basePath =
+        "/apis/spark.apache.org/v1/namespaces/"
+            + DEFAULT_NS
+            + "/sparkapplications/"
+            + testResource.getMetadata().getName();
+    var statusPath = basePath + "/status";
+    // First status update returns 409 Conflict
+    server.expect().withPath(statusPath).andReturn(409, null).once();
+    // After 409, GET latest resource with resourceVersion "2"
+    server.expect().withPath(basePath).andReturn(200, resourceV2).once();
+    // Second status update succeeds with refreshed resourceVersion
+    server.expect().withPath(statusPath).andReturn(200, resourceV3).once();
+
+    statusRecorder.persistStatus(context, new ApplicationStatus());
+
+    verify(mockStatusListener, times(1))
+        .listenStatus(
+            assertArg(a -> assertThat(a.getMetadata().getResourceVersion()).isEqualTo("3")),
+            any(),
+            any());
+  }
+
   @Test
   void retriesFailedStatusPatches() {
     var testResource = getSparkApplication("1");