feat: provide de-duplicated secondary resources stream on Context

Alternative without requiring collecting to a Map

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

Author: metacosm

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

@@ -15,6 +15,7 @@
  */
 package io.javaoperatorsdk.operator.api.reconciler;
 
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -29,6 +30,7 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.ReconcilerUtilsInternal;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedWorkflowAndDependentResourceContext;
@@ -38,8 +40,6 @@
 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);
 
@@ -84,22 +84,47 @@ public <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType, boolea
       throw new IllegalArgumentException("Can only de-duplicate HasMetadata descendants");
     }
 
+    final var idToLatest = deduplicate ? new HashMap<ResourceID, String>() : null;
     final var stream =
         controller.getEventSourceManager().getEventSourcesFor(expectedType).stream()
             .<R>mapMulti(
-                (es, consumer) -> es.getSecondaryResources(primaryResource).forEach(consumer));
+                (es, consumer) ->
+                    es.getSecondaryResources(primaryResource)
+                        .forEach(
+                            r -> {
+                              var reject = false;
+                              if (deduplicate) {
+                                final boolean[] rejectAr = new boolean[1];
+                                final var hm = (HasMetadata) r;
+                                final var resourceVersion = hm.getMetadata().getResourceVersion();
+                                idToLatest.merge(
+                                    ResourceID.fromResource(hm),
+                                    resourceVersion,
+                                    (existing, replacement) -> {
+                                      final var comparison =
+                                          ReconcilerUtilsInternal.compareResourceVersions(
+                                              existing, replacement);
+                                      rejectAr[0] =
+                                          comparison == 0; // rejecting resource if version is equal
+                                      return comparison >= 0 ? existing : replacement;
+                                    });
+                                reject = rejectAr[0];
+                              }
+                              // only keep resources that don't have the same id and resource
+                              // version
+                              if (!reject) {
+                                consumer.accept(r);
+                              }
+                            }));
     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()
+          .filter(
+              hm -> {
+                final var resourceVersion = hm.getMetadata().getResourceVersion();
+                return resourceVersion.equals(idToLatest.get(ResourceID.fromResource(hm)));
+              })
           .map(hasMetadata -> (R) hasMetadata);
     } else {
       return stream;