IGNITE-22530 WIP

Author: Andrei Nadyktov

modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractIgniteCdcStreamer.java

@@ -19,6 +19,7 @@
 
 import java.nio.file.Path;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -119,7 +120,7 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumerEx {
     }
 
     /** {@inheritDoc} */
-    @Override public void start(MetricRegistry reg, Path cdcDir) {
+    @Override public void start(MetricRegistry reg, Path cdcDir, List<String> cacheNames) {
         A.notEmpty(caches, "caches");
 
         regexManager = new CdcRegexManager(cdcDir, log);
@@ -131,6 +132,8 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumerEx {
 
         regexManager.compileRegexp(includeTemplate, excludeTemplate);
 
+        regexManager.match(cacheNames);
+
         regexManager.getSavedCaches().stream()
             .map(CU::cacheId)
             .forEach(cachesIds::add);

modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcRegexManager.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.regex.Pattern;
@@ -73,7 +74,11 @@ public CdcRegexManager(Path cdcDir, IgniteLogger log) {
      * @return True if cache name matches user's regexp patterns.
      */
     public boolean match(String cacheName) {
-        return matchAndSave(cacheName);
+        return matchAndSave(Collections.singletonList(cacheName));
+    }
+
+    public boolean match(List<String> cacheNames) {
+        return matchAndSave(cacheNames);
     }
 
     /**
@@ -98,28 +103,32 @@ public List<String> getSavedCaches() {
      * Finds match between cache name and user's regex templates.
      * If match is found, saves cache name to file.
      *
-     * @param cacheName Cache name.
+     * @param cacheNames Cache name.
      * @return True if cache name matches user's regexp patterns.
      */
-    private boolean matchAndSave(String cacheName) {
-        if (matchesFilters(cacheName)) {
-            try {
-                List<String> caches = loadCaches();
+    private boolean matchAndSave(List<String> cacheNames) {
+        try {
+            List<String> caches = loadCaches();
 
-                caches.add(cacheName);
+            caches.addAll(cacheNames);
 
-                save(caches);
-            }
-            catch (IOException e) {
-                throw new IgniteException(e);
-            }
+            save(caches);
+        }
+        catch (IOException e) {
+            throw new IgniteException(e);
+        }
 
-            if (log.isInfoEnabled())
-                log.info("Cache has been added to replication [cacheName=" + cacheName + "]");
+        List<String> matchingCaches = cacheNames.stream()
+            .filter(this::matchesFilters)
+            .collect(Collectors.toList());
 
-            return true;
-        }
-        return false;
+        if (matchingCaches.isEmpty())
+            return false;
+
+        if (log.isInfoEnabled())
+            log.info("Cache(s) has been added to replication [cacheNames=" + matchingCaches + "]");
+
+        return true;
     }
 
     /**

modules/cdc-ext/src/main/java/org/apache/ignite/cdc/IgniteToIgniteCdcStreamer.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.cdc;
 
 import java.nio.file.Path;
+import java.util.List;
 
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.Ignition;
@@ -61,8 +62,8 @@ public class IgniteToIgniteCdcStreamer extends AbstractIgniteCdcStreamer impleme
     private volatile boolean alive = true;
 
     /** {@inheritDoc} */
-    @Override public void start(MetricRegistry mreg, Path cdcDir) {
-        super.start(mreg, cdcDir);
+    @Override public void start(MetricRegistry mreg, Path cdcDir, List<String> cacheNames) {
+        super.start(mreg, cdcDir, cacheNames);
 
         if (log.isInfoEnabled())
             log.info("Ignite To Ignite Streamer [cacheIds=" + cachesIds + ']');

modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java

@@ -336,7 +336,7 @@ private <T> void sendOneBatch(
     }
 
     /** {@inheritDoc} */
-    @Override public void start(MetricRegistry reg, Path cdcDir) {
+    @Override public void start(MetricRegistry reg, Path cdcDir, List<String> cacheNames) {
         A.notNull(kafkaProps, "Kafka properties");
         A.notNull(evtTopic, "Kafka topic");
         A.notNull(metadataTopic, "Kafka metadata topic");
@@ -355,6 +355,8 @@ private <T> void sendOneBatch(
 
         regexManager.compileRegexp(includeTemplate, excludeTemplate);
 
+        regexManager.match(cacheNames);
+
         regexManager.getSavedCaches().stream()
                 .map(CU::cacheId)
                 .forEach(cachesIds::add);

modules/cdc-ext/src/main/java/org/apache/ignite/cdc/thin/IgniteToIgniteClientCdcStreamer.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.cdc.thin;
 
 import java.nio.file.Path;
+import java.util.List;
 
 import org.apache.ignite.Ignition;
 import org.apache.ignite.cdc.AbstractIgniteCdcStreamer;
@@ -68,8 +69,8 @@ public class IgniteToIgniteClientCdcStreamer extends AbstractIgniteCdcStreamer {
     private long aliveCheckTimeout = DFLT_ALIVE_CHECK_TIMEOUT;
 
     /** {@inheritDoc} */
-    @Override public void start(MetricRegistry mreg, Path cdcDir) {
-        super.start(mreg, cdcDir);
+    @Override public void start(MetricRegistry mreg, Path cdcDir, List<String> cacheNames) {
+        super.start(mreg, cdcDir, cacheNames);
 
         if (log.isInfoEnabled())
             log.info("Ignite To Ignite Client Streamer [cacheIds=" + cachesIds + ']');