Repository: kroxylicious/kroxylicious (java)

Issue:
Operator: VirtualKafkaClusterReconciler require all Referent resources to have fresh status
Following on from #2047 we want to apply the same logic in the VirtualKafkaClusterReconciler

It should:
1. ignore events from Referents with stale status, so that we don't reconcile any VKCs in response.
2. terminate VKC reconciliation if we find a stale status on any referents. We do not update the status of the VKC in this case as it should be a transient state.

Pull request description:
operator: VKC reconciler ignores events from stale referents
### Type of change

- Enhancement / new feature

### Description

When deciding if CRUD events should trigger a VirtualKafkaCluster reconcilation, we do not trigger any reconciliations if the resource has a stale status (`status.observedGeneration != metadata.generation`).

Before we run an aggregating reconciler like the VirtualKafkaCluster we want all the referents to be in a reconciled state. The aggregating reconciler can not act on a referent that has not been reconciled because the referent's reconciler may decide that there is some problem.

Closes #2069

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [ ] PR raised from a fork of this repository and made from a branch rather than main. 
- [ ] Write tests
- [ ] Update documentation
- [ ] Make sure all unit/integration tests pass
- [ ] Make sure all Sonarcloud warnings are addressed or are justifiably ignored.
- [ ] If applicable to the change, [trigger the system test suite](../blob/main/DEV_GUIDE.md#jenkins-pipeline-for-system-tests).  Make sure tests pass.
- [ ] If applicable to the change, [trigger the performance test suite](../blob/main/PERFORMANCE.md#jenkins-pipeline-for-performance). Ensure that any degradations to performance numbers are understood and justified.
- [ ] Ensure the PR references relevant issue(s) so they are closed on merging.
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).

> **_NOTE:_**  You must be a member of `@kroxylicious/developers` to trigger the system test and performance test suites.  If you are not part of this group, comment on the PR requesting a trigger, tagging `@kroxylicious/developers`.


Patch:
diff --git a/kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/ResourcesUtil.java b/kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/ResourcesUtil.java
index 81eb42e5..2dfcecd1 100644
--- a/kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/ResourcesUtil.java
+++ b/kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/ResourcesUtil.java
@@ -290,7 +290,13 @@ public class ResourcesUtil {
         return ResourcesUtil.filteredResourceIdsInSameNamespace(context,
                 referent,
                 owner,
-                primary -> refAccessor.apply(primary).stream().anyMatch(ref -> ResourcesUtil.isReferent(ref, referent)));
+                primary -> {
+                    Collection<? extends LocalRef<R>> refs = refAccessor.apply(primary);
+                    if (refs == null) {
+                        return false;
+                    }
+                    return refs.stream().anyMatch(ref -> ResourcesUtil.isReferent(ref, referent));
+                });
     }
 
     public static String namespacedSlug(LocalRef<?> ref, HasMetadata resource) {
diff --git a/kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/VirtualKafkaClusterReconciler.java b/kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/VirtualKafkaClusterReconciler.java
index 627d1680..4de87330 100644
--- a/kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/VirtualKafkaClusterReconciler.java
+++ b/kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/VirtualKafkaClusterReconciler.java
@@ -19,6 +19,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.OwnerReference;
 import io.fabric8.kubernetes.api.model.Secret;
 import io.fabric8.kubernetes.api.model.Service;
@@ -306,10 +307,7 @@ public final class VirtualKafkaClusterReconciler implements
                 .withName(SERVICES_EVENT_SOURCE_NAME)
                 .withPrimaryToSecondaryMapper((VirtualKafkaCluster cluster) -> ResourcesUtil.localRefAsResourceId(cluster,
                         cluster.getSpec().getTargetKafkaServiceRef()))
-                .withSecondaryToPrimaryMapper(service -> ResourcesUtil.findReferrers(context,
-                        service,
-                        VirtualKafkaCluster.class,
-                        cluster -> Optional.of(cluster.getSpec().getTargetKafkaServiceRef())))
+                .withSecondaryToPrimaryMapper(kafkaServiceSecondaryToPrimaryMapper(context))
                 .build();
 
         InformerEventSourceConfiguration<KafkaProxyIngress> clusterToIngresses = InformerEventSourceConfiguration.from(
@@ -321,10 +319,7 @@ public final class VirtualKafkaClusterReconciler implements
                                 .map(VirtualKafkaClusterSpec::getIngresses)
                                 .stream().flatMap(List::stream)
                                 .map(Ingresses::getIngressRef).toList()))
-                .withSecondaryToPrimaryMapper(ingress -> ResourcesUtil.findReferrersMulti(context,
-                        ingress,
-                        VirtualKafkaCluster.class,
-                        cluster -> cluster.getSpec().getIngresses().stream().map(Ingresses::getIngressRef).toList()))
+                .withSecondaryToPrimaryMapper(ingressSecondaryToPrimaryMapper(context))
                 .build();
 
         InformerEventSourceConfiguration<KafkaProtocolFilter> clusterToFilters = InformerEventSourceConfiguration.from(
@@ -333,10 +328,7 @@ public final class VirtualKafkaClusterReconciler implements
                 .withName(FILTERS_EVENT_SOURCE_NAME)
                 .withPrimaryToSecondaryMapper((VirtualKafkaCluster cluster) -> ResourcesUtil.localRefsAsResourceIds(cluster,
                         Optional.ofNullable(cluster.getSpec()).map(VirtualKafkaClusterSpec::getFilterRefs).orElse(List.of())))
-                .withSecondaryToPrimaryMapper(filter -> ResourcesUtil.findReferrersMulti(context,
-                        filter,
-                        VirtualKafkaCluster.class,
-                        cluster -> cluster.getSpec().getFilterRefs()))
+                .withSecondaryToPrimaryMapper(filterSecondaryToPrimaryMapper(context))
                 .build();
 
         InformerEventSourceConfiguration<Service> clusterToKubeService = InformerEventSourceConfiguration.from(
@@ -394,6 +386,54 @@ public final class VirtualKafkaClusterReconciler implements
                         .map(Tls::getCertificateRef).toList());
     }
 
+    @VisibleForTesting
+    static SecondaryToPrimaryMapper<KafkaProtocolFilter> filterSecondaryToPrimaryMapper(EventSourceContext<VirtualKafkaCluster> context) {
+        return filter -> {
+            if (!ResourcesUtil.isStatusFresh(filter)) {
+                logIgnoredEvent(filter);
+                return Set.of();
+            }
+            return ResourcesUtil.findReferrersMulti(context,
+                    filter,
+                    VirtualKafkaCluster.class,
+                    cluster -> cluster.getSpec().getFilterRefs());
+        };
+    }
+
+    @VisibleForTesting
+    static SecondaryToPrimaryMapper<KafkaProxyIngress> ingressSecondaryToPrimaryMapper(EventSourceContext<VirtualKafkaCluster> context) {
+        return ingress -> {
+            if (!ResourcesUtil.isStatusFresh(ingress)) {
+                logIgnoredEvent(ingress);
+                return Set.of();
+            }
+            return ResourcesUtil.findReferrersMulti(context,
+                    ingress,
+                    VirtualKafkaCluster.class,
+                    cluster -> cluster.getSpec().getIngresses().stream().map(Ingresses::getIngressRef).toList());
+        };
+    }
+
+    @VisibleForTesting
+    static SecondaryToPrimaryMapper<KafkaService> kafkaServiceSecondaryToPrimaryMapper(EventSourceContext<VirtualKafkaCluster> context) {
+        return service -> {
+            if (!ResourcesUtil.isStatusFresh(service)) {
+                logIgnoredEvent(service);
+                return Set.of();
+            }
+            return ResourcesUtil.findReferrers(context,
+                    service,
+                    VirtualKafkaCluster.class,
+                    cluster -> Optional.of(cluster.getSpec().getTargetKafkaServiceRef()));
+        };
+    }
+
+    private static void logIgnoredEvent(HasMetadata hasMetadata) {
+        if (LOGGER.isDebugEnabled()) {
+            LOGGER.debug("Ignoring event from {} with stale status: {}", HasMetadata.getKind(hasMetadata.getClass()), ResourcesUtil.toLocalRef(hasMetadata));
+        }
+    }
+
     @Override
     public ErrorStatusUpdateControl<VirtualKafkaCluster> updateErrorStatus(VirtualKafkaCluster cluster, Context<VirtualKafkaCluster> context, Exception e) {
         // ResolvedRefs to UNKNOWN


Test patch:
diff --git a/kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/ResourcesUtilTest.java b/kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/ResourcesUtilTest.java
index acced9a4..a807c1f3 100644
--- a/kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/ResourcesUtilTest.java
+++ b/kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/ResourcesUtilTest.java
@@ -156,6 +156,99 @@ class ResourcesUtilTest {
         assertThat(resources).containsExactly(ResourceID.fromResource(cm));
     }
 
+    @Test
+    void findReferrersMultiShouldMapEmptyRefsToEmptySet() {
+        // Given
+        ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "primary").endMetadata().build();
+        EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm), ConfigMap.class);
+        HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
+
+        // When
+        var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
+                configMap -> Set.of());
+
+        // Then
+        assertThat(resources).isEmpty();
+    }
+
+    @Test
+    void findReferrersMultiShouldMapSingleRefToResourceInContext() {
+        // Given
+        ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "primary").endMetadata().build();
+        EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm), ConfigMap.class);
+        HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
+
+        // When
+        var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
+                configMap -> Set.of(new AnyLocalRefBuilder().withName(configMap.getMetadata().getAnnotations().get("ref")).build()));
+
+        // Then
+        assertThat(resources).containsExactly(ResourceID.fromResource(cm));
+    }
+
+    @Test
+    void findReferrersMultiShouldMapMultipleRefsToCorrespondingResourcesInContext() {
+        // Given
+        ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "primary").endMetadata().build();
+        ConfigMap cm2 = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("bar").addToAnnotations("ref", "primary").endMetadata().build();
+        EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm, cm2), ConfigMap.class);
+        HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
+
+        // When
+        var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
+                configMap -> Set.of(new AnyLocalRefBuilder().withName(configMap.getMetadata().getAnnotations().get("ref")).build()));
+
+        // Then
+        assertThat(resources).containsExactlyInAnyOrder(ResourceID.fromResource(cm), ResourceID.fromResource(cm2));
+    }
+
+    @Test
+    void findReferrersMultiShouldExcludeResourcesThatDontReferencePrimary() {
+        // Given
+        ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "primary").endMetadata().build();
+        ConfigMap cm2 = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("bar").addToAnnotations("ref", "not-primary").endMetadata().build();
+        EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm, cm2), ConfigMap.class);
+        HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
+
+        // When
+        var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
+                configMap -> Set.of(new AnyLocalRefBuilder().withName(configMap.getMetadata().getAnnotations().get("ref")).build()));
+
+        // Then
+        assertThat(resources).containsExactlyInAnyOrder(ResourceID.fromResource(cm));
+    }
+
+    @Test
+    void findReferrersMultiShouldHandleAllSecondariesDontReferencePrimary() {
+        // Given
+        ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "not-primary").endMetadata().build();
+        ConfigMap cm2 = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("bar").addToAnnotations("ref", "not-primary").endMetadata().build();
+        EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm, cm2), ConfigMap.class);
+        HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
+
+        // When
+        var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
+                configMap -> Set.of(new AnyLocalRefBuilder().withName(configMap.getMetadata().getAnnotations().get("ref")).build()));
+
+        // Then
+        assertThat(resources).isEmpty();
+    }
+
+    @Test
+    void findReferrersMultiShouldTolerateNullReferencesCollection() {
+        // Given
+        ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "not-primary").endMetadata().build();
+        EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm), ConfigMap.class);
+        HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
+
+        // When
+        var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
+                configMap -> null);
+
+        // Then
+        assertThat(resources).isEmpty();
+    }
+
     @Test
     void findReferrersSupportsResourcesWithoutReferences() {
         // Given
diff --git a/kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/VirtualKafkaClusterReconcilerTest.java b/kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/VirtualKafkaClusterReconcilerTest.java
index 050b30bb..b2a09911 100644
--- a/kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/VirtualKafkaClusterReconcilerTest.java
+++ b/kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/VirtualKafkaClusterReconcilerTest.java
@@ -39,8 +39,11 @@ import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.SecondaryToPrimaryMapper;
 
 import io.kroxylicious.kubernetes.api.common.Condition;
+import io.kroxylicious.kubernetes.api.common.FilterRefBuilder;
+import io.kroxylicious.kubernetes.api.common.KafkaServiceRefBuilder;
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxy;
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyBuilder;
 import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyIngress;
@@ -656,12 +659,7 @@ class VirtualKafkaClusterReconcilerTest {
     @Test
     void mappingToSecretToVirtualKafkaClusterWithTls() {
         // Given
-        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mock();
-        KubernetesClient client = mock();
-        when(eventSourceContext.getClient()).thenReturn(client);
-
-        KubernetesResourceList<VirtualKafkaCluster> mockList = mockListOperation(client, VirtualKafkaCluster.class);
-        when(mockList.getItems()).thenReturn(List.of(CLUSTER_NO_FILTERS_WITH_TLS));
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mockContextContaining(CLUSTER_NO_FILTERS_WITH_TLS);
 
         // When
         var mapper = VirtualKafkaClusterReconciler.secretToVirtualKafkaCluster(eventSourceContext);
@@ -674,12 +672,7 @@ class VirtualKafkaClusterReconcilerTest {
     @Test
     void mappingToSecretToVirtualKafkaClusterToleratesVirtualKafkaClusterWithoutTls() {
         // Given
-        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mock();
-        KubernetesClient client = mock();
-        when(eventSourceContext.getClient()).thenReturn(client);
-
-        KubernetesResourceList<VirtualKafkaCluster> mockList = mockListOperation(client, VirtualKafkaCluster.class);
-        when(mockList.getItems()).thenReturn(List.of(CLUSTER_NO_FILTERS));
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mockContextContaining(CLUSTER_NO_FILTERS);
 
         // When
         var mapper = VirtualKafkaClusterReconciler.secretToVirtualKafkaCluster(eventSourceContext);
@@ -694,7 +687,171 @@ class VirtualKafkaClusterReconcilerTest {
         when(context.getSecondaryResource(Secret.class, VirtualKafkaClusterReconciler.SECRETS_EVENT_SOURCE_NAME)).thenReturn(optional);
     }
 
-    private <T extends HasMetadata> KubernetesResourceList<T> mockListOperation(KubernetesClient client, Class<T> clazz) {
+    @Test
+    void ingressSecondaryToPrimaryMapper() {
+        // given
+        VirtualKafkaCluster cluster = new VirtualKafkaClusterBuilder().withNewMetadata().withName("cluster").endMetadata().withNewSpec()
+                .withIngresses(new IngressesBuilder()
+                        .withNewIngressRef().withName("ingress").endIngressRef().build())
+                .endSpec().build();
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mockContextContaining(cluster);
+        SecondaryToPrimaryMapper<KafkaProxyIngress> mapper = VirtualKafkaClusterReconciler.ingressSecondaryToPrimaryMapper(eventSourceContext);
+        KafkaProxyIngress ingress = new KafkaProxyIngressBuilder().withNewMetadata().withName("ingress").endMetadata().withNewSpec().withNewProxyRef()
+                .withName("proxy")
+                .endProxyRef().endSpec().build();
+
+        // when
+        Set<ResourceID> primaryResourceIDs = mapper.toPrimaryResourceIDs(ingress);
+
+        // then
+        assertThat(primaryResourceIDs).containsExactly(ResourceID.fromResource(cluster));
+    }
+
+    @Test
+    void filterSecondaryToPrimaryMapper() {
+        // given
+        VirtualKafkaCluster cluster = new VirtualKafkaClusterBuilder().withNewMetadata().withName("cluster").endMetadata().withNewSpec()
+                .withFilterRefs(new FilterRefBuilder().withName("filter").build()).endSpec().build();
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mockContextContaining(cluster);
+        SecondaryToPrimaryMapper<KafkaProtocolFilter> mapper = VirtualKafkaClusterReconciler.filterSecondaryToPrimaryMapper(eventSourceContext);
+        KafkaProtocolFilter filter = new KafkaProtocolFilterBuilder().withNewMetadata().withName("filter").endMetadata().withNewSpec().endSpec().build();
+
+        // when
+        Set<ResourceID> primaryResourceIDs = mapper.toPrimaryResourceIDs(filter);
+
+        // then
+        assertThat(primaryResourceIDs).containsExactly(ResourceID.fromResource(cluster));
+    }
+
+    @Test
+    void kafkaServiceSecondaryToPrimaryMapper() {
+        // given
+        VirtualKafkaCluster cluster = new VirtualKafkaClusterBuilder().withNewMetadata().withName("cluster").endMetadata().withNewSpec()
+                .withTargetKafkaServiceRef(new KafkaServiceRefBuilder().withName("target-kafka").build()).endSpec().build();
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mockContextContaining(cluster);
+        SecondaryToPrimaryMapper<KafkaService> mapper = VirtualKafkaClusterReconciler.kafkaServiceSecondaryToPrimaryMapper(eventSourceContext);
+        KafkaService service = new KafkaServiceBuilder().withNewMetadata().withName("target-kafka").endMetadata().withNewSpec().endSpec().build();
+
+        // when
+        Set<ResourceID> primaryResourceIDs = mapper.toPrimaryResourceIDs(service);
+
+        // then
+        assertThat(primaryResourceIDs).containsExactly(ResourceID.fromResource(cluster));
+    }
+
+    @Test
+    void filterSecondaryToPrimaryMapperHandlesNullFilterRefs() {
+        // given
+        VirtualKafkaCluster clusterWithNullFilterRefs = new VirtualKafkaClusterBuilder().withNewMetadata().withName("cluster").endMetadata().withNewSpec().endSpec()
+                .build();
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mockContextContaining(clusterWithNullFilterRefs);
+        SecondaryToPrimaryMapper<KafkaProtocolFilter> mapper = VirtualKafkaClusterReconciler.filterSecondaryToPrimaryMapper(eventSourceContext);
+        KafkaProtocolFilter filter = new KafkaProtocolFilterBuilder().withNewMetadata().withName("filter").endMetadata().withNewSpec().endSpec().build();
+
+        // when
+        Set<ResourceID> primaryResourceIDs = mapper.toPrimaryResourceIDs(filter);
+
+        // then
+        assertThat(primaryResourceIDs).isEmpty();
+    }
+
+    @Test
+    void ingressSecondaryToPrimaryMapperIgnoresIngressWithStaleStatus() {
+        // given
+        VirtualKafkaCluster cluster = new VirtualKafkaClusterBuilder().withNewMetadata().withName("cluster").endMetadata().withNewSpec()
+                .withIngresses(new IngressesBuilder().withNewIngressRef().withName("ingress").endIngressRef().build()).endSpec().build();
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mockContextContaining(cluster);
+        SecondaryToPrimaryMapper<KafkaProxyIngress> mapper = VirtualKafkaClusterReconciler.ingressSecondaryToPrimaryMapper(eventSourceContext);
+        // @formatter:off
+        KafkaProxyIngress ingress = new KafkaProxyIngressBuilder()
+                .withNewMetadata()
+                .withName("ingress")
+                .withGeneration(23L)
+                .endMetadata()
+                .withNewSpec()
+                .withNewProxyRef()
+                .withName("proxy")
+                .endProxyRef()
+                .endSpec()
+                .withNewStatus()
+                .withObservedGeneration(20L)
+                .endStatus()
+                .build();
+        // @formatter:on
+
+        // when
+        Set<ResourceID> primaryResourceIDs = mapper.toPrimaryResourceIDs(ingress);
+
+        // then
+        assertThat(primaryResourceIDs).isEmpty();
+    }
+
+    @Test
+    void filterSecondaryToPrimaryMapperIgnoresFilterWithStaleStatus() {
+        // given
+        VirtualKafkaCluster cluster = new VirtualKafkaClusterBuilder().withNewMetadata().withName("cluster").endMetadata().withNewSpec()
+                .withFilterRefs(new FilterRefBuilder().withName("filter").build()).endSpec().build();
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mockContextContaining(cluster);
+        SecondaryToPrimaryMapper<KafkaProtocolFilter> mapper = VirtualKafkaClusterReconciler.filterSecondaryToPrimaryMapper(eventSourceContext);
+        // @formatter:off
+        KafkaProtocolFilter ingress = new KafkaProtocolFilterBuilder()
+                .withNewMetadata()
+                .withName("filter")
+                .withGeneration(23L)
+                .endMetadata()
+                .withNewSpec()
+                .endSpec()
+                .withNewStatus()
+                .withObservedGeneration(20L)
+                .endStatus()
+                .build();
+        // @formatter:on
+
+        // when
+        Set<ResourceID> primaryResourceIDs = mapper.toPrimaryResourceIDs(ingress);
+
+        // then
+        assertThat(primaryResourceIDs).isEmpty();
+    }
+
+    @Test
+    void kafkaServiceSecondaryToPrimaryMapperIgnoresServiceWithStaleStatus() {
+        // given
+        VirtualKafkaCluster cluster = new VirtualKafkaClusterBuilder().withNewMetadata().withName("cluster").endMetadata().withNewSpec()
+                .withTargetKafkaServiceRef(new KafkaServiceRefBuilder().withName("target-kafka").build()).endSpec().build();
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mockContextContaining(cluster);
+        SecondaryToPrimaryMapper<KafkaService> mapper = VirtualKafkaClusterReconciler.kafkaServiceSecondaryToPrimaryMapper(eventSourceContext);
+        // @formatter:off
+        KafkaService service = new KafkaServiceBuilder()
+                .withNewMetadata()
+                .withName("target-kafka")
+                .withGeneration(23L)
+                .endMetadata()
+                .withNewSpec()
+                .endSpec()
+                .withNewStatus()
+                .withObservedGeneration(20L)
+                .endStatus()
+                .build();
+        // @formatter:on
+
+        // when
+        Set<ResourceID> primaryResourceIDs = mapper.toPrimaryResourceIDs(service);
+
+        // then
+        assertThat(primaryResourceIDs).isEmpty();
+    }
+
+    private static EventSourceContext<VirtualKafkaCluster> mockContextContaining(VirtualKafkaCluster cluster) {
+        EventSourceContext<VirtualKafkaCluster> eventSourceContext = mock();
+        KubernetesClient client = mock();
+        when(eventSourceContext.getClient()).thenReturn(client);
+        KubernetesResourceList<VirtualKafkaCluster> mockList = mockListOperation(client, VirtualKafkaCluster.class);
+        when(mockList.getItems()).thenReturn(List.of(cluster));
+        return eventSourceContext;
+    }
+
+    private static <T extends HasMetadata> KubernetesResourceList<T> mockListOperation(KubernetesClient client, Class<T> clazz) {
         MixedOperation<T, KubernetesResourceList<T>, Resource<T>> mockOperation = mock();
         when(client.resources(clazz)).thenReturn(mockOperation);
         KubernetesResourceList<T> mockList = mock();
@@ -702,5 +859,4 @@ class VirtualKafkaClusterReconcilerTest {
         when(mockOperation.inNamespace(any())).thenReturn(mockOperation);
         return mockList;
     }
-
 }