IGNITE-22530 Make caches set in KafkaToIgniteCdcStreamerApplier mutable

apache · Nov 13, 2024 · 2517c25 · 2517c25
1 parent c862d18
commit 2517c25
Showing 1 changed file with 4 additions and 4 deletions.
diff --git a/...s/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/AbstractKafkaToIgniteCdcStreamer.java b/...s/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/AbstractKafkaToIgniteCdcStreamer.java
@@ -19,11 +19,11 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.cdc.AbstractCdcEventsApplier;
@@ -152,13 +152,13 @@ public AbstractKafkaToIgniteCdcStreamer(Properties kafkaProps, KafkaToIgniteCdcS
     protected void runAppliers() {
         AtomicBoolean stopped = new AtomicBoolean();
 
-        Set<Integer> caches = null;
+        Set<Integer> caches = new HashSet<>();
 
         if (!F.isEmpty(streamerCfg.getCaches())) {
             checkCaches(streamerCfg.getCaches());
 
-            caches = streamerCfg.getCaches().stream()
-                .map(CU::cacheId).collect(Collectors.toSet());
+            streamerCfg.getCaches().stream()
+                .map(CU::cacheId).forEach(caches::add);
         }
 
         KafkaToIgniteMetadataUpdater metaUpdr = new KafkaToIgniteMetadataUpdater(