feat: provide de-duplicated secondary resources stream on Context

Signed-off-by: Chris Laprun <metacosm@gmail.com>

Author: metacosm

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Context.java

@@ -38,9 +38,11 @@ default <R> Optional<R> getSecondaryResource(Class<R> expectedType) {
   <R> Set<R> getSecondaryResources(Class<R> expectedType);
 
   default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
-    return getSecondaryResources(expectedType).stream();
+    return getSecondaryResourcesAsStream(expectedType, false);
   }
 
+  <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType, boolean deduplicate);
+
   <R> Optional<R> getSecondaryResource(Class<R> expectedType, String eventSourceName);
 
   ControllerConfiguration<P> getControllerConfiguration();

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java

@@ -24,6 +24,9 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
@@ -35,7 +38,10 @@
 import io.javaoperatorsdk.operator.processing.event.NoEventSourceForClassException;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
+import static io.javaoperatorsdk.operator.ReconcilerUtilsInternal.compareResourceVersions;
+
 public class DefaultContext<P extends HasMetadata> implements Context<P> {
+  private static final Logger log = LoggerFactory.getLogger(DefaultContext.class);
 
   private RetryInfo retryInfo;
   private final Controller<P> controller;
@@ -73,11 +79,31 @@ public <T> Set<T> getSecondaryResources(Class<T> expectedType) {
     return getSecondaryResourcesAsStream(expectedType).collect(Collectors.toSet());
   }
 
-  @Override
-  public <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
-    return controller.getEventSourceManager().getEventSourcesFor(expectedType).stream()
-        .map(es -> es.getSecondaryResources(primaryResource))
-        .flatMap(Set::stream);
+  public <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType, boolean deduplicate) {
+    if (deduplicate && !HasMetadata.class.isAssignableFrom(expectedType)) {
+      throw new IllegalArgumentException("Can only de-duplicate HasMetadata descendants");
+    }
+
+    final var stream =
+        controller.getEventSourceManager().getEventSourcesFor(expectedType).stream()
+            .<R>mapMulti(
+                (es, consumer) -> es.getSecondaryResources(primaryResource).forEach(consumer));
+    if (deduplicate) {
+      //noinspection unchecked
+      return stream
+          .map(HasMetadata.class::cast)
+          .collect(
+              Collectors.toUnmodifiableMap(
+                  ResourceID::fromResource,
+                  Function.identity(),
+                  (existing, replacement) ->
+                      compareResourceVersions(existing, replacement) >= 0 ? existing : replacement))
+          .values()
+          .stream()
+          .map(hasMetadata -> (R) hasMetadata);
+    } else {
+      return stream;
+    }
   }
 
   @Override

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ReconcileUtils.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.UnaryOperator;
 import java.util.stream.Collector;
@@ -650,9 +651,8 @@ public static <P extends HasMetadata> P addFinalizerWithSSA(
 
   private static <T extends HasMetadata> Collector<T, ?, Map<ResourceID, T>> latestDistinctToMap() {
     return Collectors.toMap(
-        resource ->
-            new ResourceID(resource.getMetadata().getName(), resource.getMetadata().getNamespace()),
-        resource -> resource,
+        ResourceID::fromResource,
+        Function.identity(),
         (existing, replacement) ->
             compareResourceVersions(existing, replacement) >= 0 ? existing : replacement);
   }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java

@@ -218,7 +218,8 @@ public Set<R> getSecondaryResources(P primary) {
     }
     return secondaryIDs.stream()
         .map(this::get)
-        .flatMap(Optional::stream)
+        .filter(Optional::isPresent)
+        .map(Optional::get)
         .collect(Collectors.toSet());
   }
 

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/latestdistinct/LatestDistinctTestReconciler.java

@@ -58,10 +58,7 @@ public UpdateControl<LatestDistinctTestResource> reconcile(
       return UpdateControl.noUpdate();
     }
     // makes sure that distinc config maps returned
-    var distinctConfigMaps =
-        context
-            .getSecondaryResourcesAsStream(ConfigMap.class)
-            .collect(ReconcileUtils.latestDistinctList());
+    var distinctConfigMaps = context.getSecondaryResourcesAsStream(ConfigMap.class, true).toList();
     if (distinctConfigMaps.size() != 1) {
       errorOccurred = true;
       throw new IllegalStateException();