Include KafkaProtocolFilter in integration tests

Filters are what we're all about! Lets add them to the integration
tests.

Resolves kroxylicious#1968

Signed-off-by: Robert Young <[email protected]>

Author: robobario

kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/ProxyReconcilerIT.java

@@ -8,8 +8,13 @@
 
 import java.time.Duration;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
+import io.kroxylicious.kubernetes.filter.api.v1alpha1.KafkaProtocolFilter;
+
+import io.kroxylicious.kubernetes.filter.api.v1alpha1.KafkaProtocolFilterBuilder;
+
 import org.assertj.core.api.AbstractStringAssert;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.awaitility.core.ConditionFactory;
@@ -56,6 +61,7 @@ class ProxyReconcilerIT {
     private static final String PROXY_A = "proxy-a";
     private static final String PROXY_B = "proxy-b";
     private static final String CLUSTER_FOO_REF = "fooref";
+    private static final String FILTER_NAME = "validation";
     private static final String CLUSTER_FOO = "foo";
     private static final String CLUSTER_FOO_CLUSTERIP_INGRESS = "foo-cluster-ip";
     private static final String CLUSTER_FOO_BOOTSTRAP = "my-cluster-kafka-bootstrap.foo.svc.cluster.local:9092";
@@ -88,6 +94,7 @@ void beforeEach() {
             .withAdditionalCustomResourceDefinition(VirtualKafkaCluster.class)
             .withAdditionalCustomResourceDefinition(KafkaService.class)
             .withAdditionalCustomResourceDefinition(KafkaProxyIngress.class)
+            .withAdditionalCustomResourceDefinition(KafkaProtocolFilter.class)
             .waitForNamespaceDeletion(true)
             .withConfigurationService(x -> x.withCloseClientOnStop(false))
             .build();
@@ -119,12 +126,13 @@ public KafkaProxyIngress ingress(String name) {
 
     CreatedResources doCreate() {
         KafkaProxy proxy = extension.create(kafkaProxy(PROXY_A));
+        KafkaProtocolFilter filter = extension.create(filter(FILTER_NAME));
         KafkaService barClusterRef = extension.create(clusterRef(CLUSTER_BAR_REF, CLUSTER_BAR_BOOTSTRAP));
         KafkaProxyIngress ingressBar = extension.create(clusterIpIngress(CLUSTER_BAR_CLUSTERIP_INGRESS, proxy));
         Set<KafkaService> clusterRefs = Set.of(barClusterRef);
-        VirtualKafkaCluster clusterBar = extension.create(virtualKafkaCluster(CLUSTER_BAR, proxy, barClusterRef, ingressBar));
+        VirtualKafkaCluster clusterBar = extension.create(virtualKafkaCluster(CLUSTER_BAR, proxy, barClusterRef, ingressBar, filter));
         Set<VirtualKafkaCluster> clusters = Set.of(clusterBar);
-        assertProxyConfigContents(proxy, Set.of(CLUSTER_BAR_BOOTSTRAP), Set.of());
+        assertProxyConfigContents(proxy, Set.of(CLUSTER_BAR_BOOTSTRAP, filter.getSpec().getType()), Set.of());
         assertDeploymentMountsConfigConfigMap(proxy);
         assertDeploymentBecomesReady(proxy);
         assertServiceTargetsProxyInstances(proxy, clusterBar, ingressBar);
@@ -274,9 +282,10 @@ void moveVirtualKafkaClusterToAnotherKafkaProxy() {
 
         KafkaService fooClusterRef = extension.create(clusterRef(CLUSTER_FOO_REF, CLUSTER_FOO_BOOTSTRAP));
         KafkaService barClusterRef = extension.create(clusterRef(CLUSTER_BAR_REF, CLUSTER_BAR_BOOTSTRAP));
+        KafkaProtocolFilter filter = extension.create(filter(FILTER_NAME));
 
-        VirtualKafkaCluster clusterFoo = extension.create(virtualKafkaCluster(CLUSTER_FOO, proxyA, fooClusterRef, ingressFoo));
-        VirtualKafkaCluster barCluster = extension.create(virtualKafkaCluster(CLUSTER_BAR, proxyB, barClusterRef, ingressBar));
+        VirtualKafkaCluster clusterFoo = extension.create(virtualKafkaCluster(CLUSTER_FOO, proxyA, fooClusterRef, ingressFoo, filter));
+        VirtualKafkaCluster barCluster = extension.create(virtualKafkaCluster(CLUSTER_BAR, proxyB, barClusterRef, ingressBar, filter));
 
         assertProxyConfigContents(proxyA, Set.of(CLUSTER_FOO_BOOTSTRAP), Set.of());
         assertProxyConfigContents(proxyB, Set.of(CLUSTER_BAR_BOOTSTRAP), Set.of());
@@ -312,13 +321,13 @@ private AbstractStringAssert<?> assertThatProxyConfigFor(KafkaProxy proxy) {
     }
 
     private static VirtualKafkaCluster virtualKafkaCluster(String clusterName, KafkaProxy proxy, KafkaService clusterRef,
-                                                           KafkaProxyIngress ingress) {
+                                                           KafkaProxyIngress ingress, KafkaProtocolFilter filter) {
         return new VirtualKafkaClusterBuilder().withNewMetadata().withName(clusterName).endMetadata()
                 .withNewSpec()
                 .withTargetKafkaServiceRef(new KafkaServiceRefBuilder().withName(name(clusterRef)).build())
                 .withNewProxyRef().withName(name(proxy)).endProxyRef()
                 .addNewIngressRef().withName(name(ingress)).endIngressRef()
-                .withFilterRefs()
+                .withFilterRefs().addNewFilterRef().withName(name(filter)).endFilterRef()
                 .endSpec().build();
     }
 
@@ -329,6 +338,12 @@ private static KafkaService clusterRef(String clusterRefName, String clusterBoot
                 .endSpec().build();
     }
 
+    private static KafkaProtocolFilter filter(String name) {
+        return new KafkaProtocolFilterBuilder().withNewMetadata().withName(name).endMetadata()
+                .withNewSpec().withType("RecordValidation").withConfigTemplate(Map.of("rules", List.of(Map.of("allowNulls", false))))
+                .endSpec().build();
+    }
+
     KafkaProxy kafkaProxy(String name) {
         // @formatter:off
         return new KafkaProxyBuilder()