parallelize InformerManager changeNamespaces (#3430)

* parallelize InformerManager changeNamespaces

InformerManager#changeNamespaces starts informers for newly added
namespaces sequentially via Set#forEach. Each source.start() call
blocks up to cacheSyncTimeout waiting for cache sync, so when N
namespaces are added and any of them are slow or RBAC-denied (a
common case under stopOnInformerErrorDuringStartup=false), the
total wall-clock is O(N * cacheSyncTimeout). The Controller holds
its EventProcessor stopped across this entire call, dropping watch
events with "Skipping event ... because the event processor is not
started" and producing multi-minute reconcile delays after dynamic
namespace updates.

Align changeNamespaces with the existing parallel pattern already
used by InformerManager#start (boundedExecuteAndWaitForAllToComplete
on the caching executor). Wall-clock drops from O(N * timeout) to
O(timeout). The sources map is already a ConcurrentHashMap, so no
additional synchronization is required.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Qi Tan <16416018+TQJADE@users.noreply.github.com>

Author: TQJADE

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

@@ -113,14 +113,24 @@ public void changeNamespaces(Set<String> namespaces) {
     log.debug("Stopped informer {} for namespaces: {}", this, sourcesToRemove);
     sourcesToRemove.forEach(k -> sources.remove(k).stop());
 
-    namespaces.forEach(
-        ns -> {
-          if (!sources.containsKey(ns)) {
-            final InformerWrapper<R> source = createEventSourceForNamespace(ns);
-            source.start();
-            log.debug("Registered new {} -> {} for namespace: {}", this, source, ns);
-          }
-        });
+    var newNamespaces =
+        namespaces.stream().filter(ns -> !sources.containsKey(ns)).collect(Collectors.toList());
+    if (newNamespaces.isEmpty()) {
+      return;
+    }
+
+    controllerConfiguration
+        .getConfigurationService()
+        .getExecutorServiceManager()
+        .boundedExecuteAndWaitForAllToComplete(
+            newNamespaces.stream(),
+            ns -> {
+              final var source = createEventSourceForNamespace(ns);
+              source.start();
+              log.debug("Registered new {} -> {} for namespace: {}", this, source, ns);
+              return null;
+            },
+            ns -> "InformerStarter-" + ns + "-" + configuration.getResourceClass().getSimpleName());
   }
 
   private InformerWrapper<R> createEventSourceForNamespace(String namespace) {