fix(tests): stablise sporadically failing OperatorChangeDetectionST (kroxylicious#3215)

why: the test's attempt to update the CR raced with the operator's update to the same resource's status section. This meant the optimistic lock (resourceVersion) used during the update could fail with Conflict, leading to the sporadically failing test.

Refactoring the test to use replaceResourceWithRetries rather than createOrUpdateResourceWithWait with should resolve the issue. This is already the approach taken by several other tests in the same suite.

Signed-off-by: Keith Wall <kwall@apache.org>

Author: k-wall

kroxylicious-systemtests/src/test/java/io/kroxylicious/systemtests/OperatorChangeDetectionST.java

@@ -42,14 +42,13 @@
 import io.kroxylicious.kubernetes.api.common.FilterRef;
 import io.kroxylicious.kubernetes.api.common.FilterRefBuilder;
 import io.kroxylicious.kubernetes.api.common.Protocol;
-import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProtocolFilter;
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProtocolFilterBuilder;
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxy;
-import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyBuilder;
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyIngress;
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyIngressBuilder;
 import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
 import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaClusterBuilder;
+import io.kroxylicious.kubernetes.api.v1alpha1.kafkaproxyspec.InfrastructureBuilder;
 import io.kroxylicious.kubernetes.operator.assertj.OperatorAssertions;
 import io.kroxylicious.systemtests.installation.kroxylicious.CertManager;
 import io.kroxylicious.systemtests.installation.kroxylicious.Kroxylicious;
@@ -184,31 +183,23 @@ void shouldUpdateDeploymentWhenDownstreamTrustUpdated(String namespace) {
     void shouldUpdateWhenFilterConfigurationChanges(String namespace) {
         // Given
         // @formatter:off
-        KafkaProtocolFilterBuilder arbitraryFilter = KroxyliciousFilterTemplates.baseFilterDeployment(namespace, "arbitrary-filter")
+        var arbitraryFilterBuilder = KroxyliciousFilterTemplates.baseFilterDeployment(namespace, "arbitrary-filter")
                 .withNewSpec()
                     .withType("io.kroxylicious.filter.simpletransform.ProduceRequestTransformation")
                     .withConfigTemplate(Map.of("transformation", "Replacing", "transformationConfig",  Map.of("findPattern", "foo", "replacementValue", "bar")))
                 .endSpec();
         // @formatter:on
-        KubeClient kubeClient = kubeClient(namespace);
-        resourceManager.createOrUpdateResourceWithWait(arbitraryFilter);
+        resourceManager.createOrUpdateResourceWithWait(arbitraryFilterBuilder);
         kroxylicious.deployPortIdentifiesNodeWithFilters(kafkaClusterName, List.of("arbitrary-filter"));
 
-        String originalChecksum = getInitialChecksum(namespace);
-
-        // @formatter:off
-        KafkaProtocolFilterBuilder updatedProtocolFilter = kubeClient.getClient().resources(KafkaProtocolFilter.class)
-                .inNamespace(namespace)
-                .withName("arbitrary-filter")
-                .get()
-                .edit()
-                    .editSpec()
-                    .withConfigTemplate(Map.of("transformation", "Replacing", "transformationConfig",  Map.of("findPattern", "foo", "replacementValue", "updated")))
-                .endSpec();
-        // @formatter:on
+        var originalChecksum = getInitialChecksum(namespace);
+        var arbitraryFilter = arbitraryFilterBuilder.build();
+        var replacementConfig = Map.of("transformation", "Replacing", "transformationConfig", Map.of("findPattern", "foo", "replacementValue", "updated"));
 
         // When
-        resourceManager.createOrUpdateResourceWithWait(updatedProtocolFilter);
+        resourceManager.replaceResourceWithRetries(arbitraryFilter, current -> {
+            current.getSpec().setConfigTemplate(replacementConfig);
+        });
         LOGGER.info("Kafka proxy filter updated");
 
         // Then
@@ -227,21 +218,20 @@ void shouldUpdateDeploymentWhenResourceRequirementsChange(String namespace) {
         var customResourceRequests = Map.of("cpu", Quantity.parse("599m"), "memory", Quantity.parse("515Mi"));
 
         // @formatter:off
-        KafkaProxyBuilder updatedKafkaProxy = kafkaProxy.edit()
-                .editOrNewSpec()
-                .withNewInfrastructure()
+        var infra = new InfrastructureBuilder()
                 .withNewProxyContainer()
                 .withResources(new ResourceRequirementsBuilder()
-                        .withLimits(customResourceLimits)
-                        .withRequests(customResourceRequests)
-                        .build())
-                .endProxyContainer()
-                .endInfrastructure()
-                .endSpec();
+                    .withLimits(customResourceLimits)
+                    .withRequests(customResourceRequests)
+                    .build())
+            .endProxyContainer()
+            .build();
         // @formatter:on
 
         // When
-        resourceManager.createOrUpdateResourceWithWait(updatedKafkaProxy);
+        resourceManager.replaceResourceWithRetries(kafkaProxy, current -> {
+            current.getSpec().setInfrastructure(infra);
+        });
         LOGGER.info("Kafka proxy updated");
 
         // Then
@@ -275,15 +265,10 @@ void shouldNotUpdateDeploymentChecksumWhenKafkaProxyScaled(String namespace) {
         KafkaProxy kafkaProxy = kubeClient.getClient().resources(KafkaProxy.class).inNamespace(namespace)
                 .withName(Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME).get();
 
-        // @formatter:off
-        KafkaProxyBuilder updatedKafkaProxy = kafkaProxy.edit()
-                .editOrNewSpec()
-                .withReplicas(2)
-                .endSpec();
-        // @formatter:on
-
         // When
-        resourceManager.createOrUpdateResourceWithWait(updatedKafkaProxy);
+        resourceManager.replaceResourceWithRetries(kafkaProxy, current -> {
+            current.getSpec().setReplicas(2);
+        });
         LOGGER.info("Kafka proxy updated");
 
         // Then