fix: re-fetch resources before status patch to avoid 409 Conflict

When running operator ITs against a remote cluster, the operator can
reconcile between create() and patchStatus(), bumping resourceVersion
and making the held object stale. Apply the same re-fetch pattern
(already used for KafkaService in KafkaProxyReconcilerIT) to all
remaining updateStatusObservedGeneration overloads in both
KafkaProxyReconcilerIT and VirtualKafkaClusterReconcilerIT.

Assisted-by: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Sam Barker <sam@quadrocket.co.uk>

Author: SamBarker

kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/reconciler/kafkaproxy/KafkaProxyReconcilerIT.java

@@ -19,9 +19,6 @@
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import org.assertj.core.api.AbstractStringAssert;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.awaitility.core.ConditionFactory;
@@ -31,6 +28,8 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 
@@ -1459,30 +1458,36 @@ void deleteAndRestoreADependency() {
 
     // the KafkaProxyReconciler only operates on Clusters that have been reconciled, ie metadata.status == status.observedGeneration
     private VirtualKafkaCluster updateStatusObservedGeneration(VirtualKafkaCluster clusterBar) {
-        clusterBar.setStatus(new VirtualKafkaClusterStatusBuilder().withObservedGeneration(generation(clusterBar)).build());
-        return operator.patchStatus(clusterBar);
+        // Re-fetch to get the latest resourceVersion - the operator may have reconciled since we created it
+        VirtualKafkaCluster fresh = Objects.requireNonNull(operator.get(VirtualKafkaCluster.class, name(clusterBar)));
+        fresh.setStatus(new VirtualKafkaClusterStatusBuilder().withObservedGeneration(generation(fresh)).build());
+        return operator.patchStatus(fresh);
     }
 
     // the KafkaProxyReconciler only operates on KafkaProtocolFilters that have been reconciled, ie metadata.status == status.observedGeneration
     private KafkaProtocolFilter updateStatusObservedGeneration(KafkaProtocolFilter filter) {
-        filter.setStatus(new KafkaProtocolFilterStatusBuilder().withObservedGeneration(generation(filter)).build());
-        return operator.patchStatus(filter);
+        // Re-fetch to get the latest resourceVersion - the operator may have reconciled since we created it
+        KafkaProtocolFilter fresh = Objects.requireNonNull(operator.get(KafkaProtocolFilter.class, name(filter)));
+        fresh.setStatus(new KafkaProtocolFilterStatusBuilder().withObservedGeneration(generation(fresh)).build());
+        return operator.patchStatus(fresh);
     }
 
     // the KafkaProxyReconciler only operates on KafkaServices that have been reconciled, ie metadata.status == status.observedGeneration
     private KafkaService updateStatusObservedGeneration(KafkaService service, String bootstrapServers) {
         // Re-fetch to get the latest resourceVersion - the operator may have reconciled since we created it
-        KafkaService fresh = operator.get(KafkaService.class, name(service));
+        KafkaService fresh = Objects.requireNonNull(operator.get(KafkaService.class, name(service)));
         fresh.setStatus(new KafkaServiceStatusBuilder().withObservedGeneration(generation(fresh))
                 .withBootstrapServers(bootstrapServers)
                 .build());
         return operator.patchStatus(fresh);
     }
 
-    // the KafkaProxyReconciler only operates on KafkaServices that have been reconciled, ie metadata.status == status.observedGeneration
+    // the KafkaProxyReconciler only operates on KafkaProxyIngresses that have been reconciled, ie metadata.status == status.observedGeneration
     private KafkaProxyIngress updateStatusObservedGeneration(KafkaProxyIngress ingress) {
-        ingress.setStatus(new KafkaProxyIngressStatusBuilder().withObservedGeneration(generation(ingress)).build());
-        return operator.patchStatus(ingress);
+        // Re-fetch to get the latest resourceVersion - the operator may have reconciled since we created it
+        KafkaProxyIngress fresh = Objects.requireNonNull(operator.get(KafkaProxyIngress.class, name(ingress)));
+        fresh.setStatus(new KafkaProxyIngressStatusBuilder().withObservedGeneration(generation(fresh)).build());
+        return operator.patchStatus(fresh);
     }
 
     private void assertDeploymentIsRemoved(KafkaProxy proxy) {

kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/reconciler/virtualkafkacluster/VirtualKafkaClusterReconcilerIT.java

@@ -10,6 +10,7 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.awaitility.core.ConditionFactory;
@@ -53,6 +54,7 @@
 import static io.kroxylicious.kubernetes.api.common.Protocol.TCP;
 import static io.kroxylicious.kubernetes.api.common.Protocol.TLS;
 import static io.kroxylicious.kubernetes.operator.ResourcesUtil.generation;
+import static io.kroxylicious.kubernetes.operator.ResourcesUtil.name;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
@@ -618,24 +620,30 @@ private static KafkaProtocolFilter filter(String name) {
         // @formatter:on
     }
 
-    // the KafkaProxyReconciler only operates on KafkaProtocolFilters that have been reconciled, ie metadata.status == status.observedGeneration
+    // the VirtualKafkaClusterReconciler only operates on KafkaProtocolFilters that have been reconciled, ie metadata.status == status.observedGeneration
     private KafkaProtocolFilter updateStatusObservedGeneration(KafkaProtocolFilter filter) {
-        filter.setStatus(new KafkaProtocolFilterStatusBuilder().withObservedGeneration(generation(filter)).build());
-        return operator.patchStatus(filter);
+        // Re-fetch to get the latest resourceVersion - the operator may have reconciled since we created it
+        KafkaProtocolFilter fresh = Objects.requireNonNull(operator.get(KafkaProtocolFilter.class, name(filter)));
+        fresh.setStatus(new KafkaProtocolFilterStatusBuilder().withObservedGeneration(generation(fresh)).build());
+        return operator.patchStatus(fresh);
     }
 
-    // the KafkaProxyReconciler only operates on KafkaServices that have been reconciled, ie metadata.status == status.observedGeneration
-    private KafkaService updateStatusObservedGeneration(KafkaService filter, String bootstrapServers) {
-        filter.setStatus(new KafkaServiceStatusBuilder().withObservedGeneration(generation(filter))
+    // the VirtualKafkaClusterReconciler only operates on KafkaServices that have been reconciled, ie metadata.status == status.observedGeneration
+    private KafkaService updateStatusObservedGeneration(KafkaService service, String bootstrapServers) {
+        // Re-fetch to get the latest resourceVersion - the operator may have reconciled since we created it
+        KafkaService fresh = Objects.requireNonNull(operator.get(KafkaService.class, name(service)));
+        fresh.setStatus(new KafkaServiceStatusBuilder().withObservedGeneration(generation(fresh))
                 .withBootstrapServers(bootstrapServers)
                 .build());
-        return operator.patchStatus(filter);
+        return operator.patchStatus(fresh);
     }
 
-    // the KafkaProxyReconciler only operates on KafkaServices that have been reconciled, ie metadata.status == status.observedGeneration
+    // the VirtualKafkaClusterReconciler only operates on KafkaProxyIngresses that have been reconciled, ie metadata.status == status.observedGeneration
     private KafkaProxyIngress updateStatusObservedGeneration(KafkaProxyIngress ingress) {
-        ingress.setStatus(new KafkaProxyIngressStatusBuilder().withObservedGeneration(generation(ingress)).build());
-        return operator.patchStatus(ingress);
+        // Re-fetch to get the latest resourceVersion - the operator may have reconciled since we created it
+        KafkaProxyIngress fresh = Objects.requireNonNull(operator.get(KafkaProxyIngress.class, name(ingress)));
+        fresh.setStatus(new KafkaProxyIngressStatusBuilder().withObservedGeneration(generation(fresh)).build());
+        return operator.patchStatus(fresh);
     }
 
 }