IGNITE-22530 Add already existing caches to CDC after new regex filters are set

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,13 +74,23 @@ 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));
     }
 
     /**
-     * Get actual list of names of caches added by regex templates from cache list file.
-     * Caches that added to replication through regex templates during the work of CDC application,
-     * are saved to file so they can be restored after application restart.
+     * Finds and processes match between a set of cache names and user's regexp patterns.
+     *
+     * @param cacheNames Cache names.
+     * @return True if cache name matches user's regexp patterns.
+     */
+    public boolean match(List<String> cacheNames) {
+        return matchAndSave(cacheNames);
+    }
+
+    /**
+     * Get actual list of names of caches added by regex templates from cache list file.<br/>
+     * All new caches started during the work of CDC application are saved to file so they can be added to CDC later if
+     * appropriate regex filter is set.
      *
      * @return Caches names list.
      */
@@ -95,31 +106,34 @@ public List<String> getSavedCaches() {
     }
 
     /**
-     * Finds match between cache name and user's regex templates.
-     * If match is found, saves cache name to file.
+     * Finds match between caches names and user's regex templates and saves cache name to a file.
      *
-     * @param cacheName Cache name.
+     * @param cacheNames Cache names.
      * @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 + ']');