Merge pull request kroxylicious#2236 from k-wall/issue-2235

Fix kroxylicious#2235: Avoid potential for NPE if virtual cluster defines TLS without trust anchor

Author: SamBarker

kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/VirtualKafkaClusterReconciler.java

@@ -466,6 +466,7 @@ static PrimaryToSecondaryMapper<VirtualKafkaCluster> virtualKafkaClusterToConfig
                 cluster.getSpec().getIngresses().stream()
                         .flatMap(ingress -> Optional.ofNullable(ingress.getTls()).stream())
                         .map(Tls::getTrustAnchorRef)
+                        .filter(Objects::nonNull)
                         .map(TrustAnchorRef::getRef)
                         .toList());
     }
@@ -478,6 +479,7 @@ static SecondaryToPrimaryMapper<ConfigMap> configMapToVirtualKafkaCluster(EventS
                 cluster -> cluster.getSpec().getIngresses().stream()
                         .flatMap(ingress -> Optional.ofNullable(ingress.getTls()).stream())
                         .map(Tls::getTrustAnchorRef)
+                        .filter(Objects::nonNull)
                         .map(TrustAnchorRef::getRef)
                         .toList());
     }

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

@@ -764,7 +764,7 @@ void canMapFromVirtualKafkaClusterWithTrustAnchorToConfigMap() {
     }
 
     @Test
-    void canMapFromVirtualKafkaClusterWithoutTrustAnchorToConfigMap() {
+    void canMapFromVirtualKafkaClusterWithTlsToConfigMap() {
         // Given
         var mapper = VirtualKafkaClusterReconciler.virtualKafkaClusterToConfigMap();
 
@@ -775,6 +775,18 @@ void canMapFromVirtualKafkaClusterWithoutTrustAnchorToConfigMap() {
         assertThat(secondaryResourceIDs).isEmpty();
     }
 
+    @Test
+    void canMapFromVirtualKafkaClusterWithoutTrustAnchorToConfigMap() {
+        // Given
+        var mapper = VirtualKafkaClusterReconciler.virtualKafkaClusterToConfigMap();
+
+        // When
+        var secondaryResourceIDs = mapper.toSecondaryResourceIDs(CLUSTER_TLS_NO_FILTERS);
+
+        // Then
+        assertThat(secondaryResourceIDs).isEmpty();
+    }
+
     @Test
     void canMapFromConfigMapToVirtualKafkaClusterWithTls() {
         // Given
@@ -801,6 +813,19 @@ void canMapFromConfigMapToVirtualKafkaClusterToleratesVirtualKafkaClusterWithout
         assertThat(primaryResourceIDs).isEmpty();
     }
 
+    @Test
+    void canMapFromConfigMapToVirtualKafkaClusterToleratesVirtualKafkaClusterWithoutTrustAnchor() {
+        // Given
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mockContextContaining(CLUSTER_TLS_NO_FILTERS);
+
+        // When
+        var mapper = VirtualKafkaClusterReconciler.configMapToVirtualKafkaCluster(eventSourceContext);
+
+        // Then
+        var primaryResourceIDs = mapper.toPrimaryResourceIDs(PEM_CONFIG_MAP);
+        assertThat(primaryResourceIDs).isEmpty();
+    }
+
     @Test
     void ingressSecondaryToPrimaryMapper() {
         // given