IGNITE-22530 Make CdcConsumerEx accept list of actual caches names

Author: Andrei Nadyktov

modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcConsumerEx.java

@@ -18,19 +18,21 @@
 package org.apache.ignite.internal.cdc;
 
 import java.nio.file.Path;
+import java.util.List;
 
 import org.apache.ignite.cdc.CdcConsumer;
 import org.apache.ignite.metric.MetricRegistry;
 
 /**
- * Extended CdcConsumer interface which provides overloaded {@link CdcConsumerEx#start(MetricRegistry, Path)} method
+ * Extended CdcConsumer interface which provides overloaded {@link CdcConsumerEx#start(MetricRegistry, Path, List)} method
  * required for CDC regex filters.
  */
 public interface CdcConsumerEx extends CdcConsumer {
     /**
      * Starts the consumer.
      * @param mreg Metric registry for consumer specific metrics.
      * @param cdcDir Path to Change Data Capture Directory.
+     * @param cacheNames List of cache names.
      */
-    void start(MetricRegistry mreg, Path cdcDir);
+    void start(MetricRegistry mreg, Path cdcDir, List<String> cacheNames);
 }

modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java

@@ -26,11 +26,13 @@
 import java.util.Comparator;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
@@ -41,6 +43,7 @@
 import org.apache.ignite.cdc.CdcConsumer;
 import org.apache.ignite.cdc.CdcEvent;
 import org.apache.ignite.cdc.TypeMapping;
+import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.DataRegionConfiguration;
 import org.apache.ignite.configuration.DataStorageConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
@@ -339,7 +342,17 @@ public void runX() throws Exception {
                     committedSegmentOffset.value(walState.get1().fileOffset());
                 }
 
-                consumer.start(mreg, kctx.metric().registry(metricName("cdc", "consumer")), ft.walCdc().toPath());
+                List<String> cacheNames = GridLocalConfigManager
+                    .readCachesData(
+                        ft,
+                        kctx.marshallerContext().jdkMarshaller(),
+                        igniteCfg)
+                    .values().stream()
+                    .map(data -> data.configuration().getName())
+                    .collect(Collectors.toList());
+
+                consumer.start(mreg, kctx.metric().registry(metricName("cdc", "consumer")), ft.walCdc().toPath(),
+                    cacheNames);
 
                 started = true;
 

modules/core/src/main/java/org/apache/ignite/internal/cdc/WalRecordsConsumer.java

@@ -20,6 +20,7 @@
 import java.nio.file.Path;
 import java.util.EnumSet;
 import java.util.Iterator;
+import java.util.List;
 import java.util.NoSuchElementException;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
@@ -188,11 +189,12 @@ public void onCacheDestroyEvents(Iterator<Integer> caches) {
      * @param cdcReg CDC metric registry.
      * @param cdcConsumerReg CDC consumer metric registry.
      * @param cdcDir Path to Change Data Capture Directory.
+     * @param cacheNames List of cache names.
      * @throws IgniteCheckedException If failed.
      */
-    public void start(MetricRegistryImpl cdcReg, MetricRegistryImpl cdcConsumerReg, Path cdcDir) throws IgniteCheckedException {
+    public void start(MetricRegistryImpl cdcReg, MetricRegistryImpl cdcConsumerReg, Path cdcDir, List<String> cacheNames) throws IgniteCheckedException {
         if (consumer instanceof CdcConsumerEx)
-            ((CdcConsumerEx) consumer).start(cdcConsumerReg, cdcDir);
+            ((CdcConsumerEx) consumer).start(cdcConsumerReg, cdcDir, cacheNames);
         else
             consumer.start(cdcConsumerReg);
 
@@ -205,7 +207,7 @@ public void start(MetricRegistryImpl cdcReg, MetricRegistryImpl cdcConsumerReg,
 
     /**
      * Stops the consumer.
-     * This methods can be invoked only after {@link #start(MetricRegistryImpl, MetricRegistryImpl, Path)}.
+     * This methods can be invoked only after {@link #start(MetricRegistryImpl, MetricRegistryImpl, Path, List)}.
      */
     public void stop() {
         consumer.stop();