IGNITE-22530 CDC: Add regex filters for cache names

Author: Andrei Nadyktov

modules/control-utility/src/test/java/org/apache/ignite/util/CdcCommandTest.java

@@ -19,6 +19,7 @@
 
 import java.io.File;
 import java.io.Serializable;
+import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
@@ -194,7 +195,7 @@ public void testDeleteLostSegmentLinksApplicationNotClosed() throws Exception {
         CdcConfiguration cfg = new CdcConfiguration();
 
         cfg.setConsumer(new UserCdcConsumer() {
-            @Override public void start(MetricRegistry mreg) {
+            @Override public void start(MetricRegistry mreg, Path cdcDir) {
                 appStarted.countDown();
             }
         });

modules/control-utility/src/test/java/org/apache/ignite/util/CdcResendCommandTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.util;
 
+import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -217,7 +218,7 @@ synchronized List<CdcEvent> events() {
         }
 
         /** {@inheritDoc} */
-        @Override public void start(MetricRegistry mreg) {
+        @Override public void start(MetricRegistry mreg, Path cdcDir) {
             // No-op
         }
 

modules/core/src/main/java/org/apache/ignite/cdc/CdcConsumer.java

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.cdc;
 
+import java.nio.file.Path;
 import java.util.Iterator;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteBinary;
@@ -33,7 +34,7 @@
  * This consumer will receive data change events during ignite-cdc process invocation.
  * The lifecycle of the consumer is the following:
  * <ul>
- *     <li>Start of the consumer {@link #start(MetricRegistry)}.</li>
+ *     <li>Start of the consumer {@link #start(MetricRegistry, Path)}.</li>
  *     <li>Notification of the consumer by the {@link #onEvents(Iterator)} call.</li>
  *     <li>Stop of the consumer {@link #stop()}.</li>
  * </ul>
@@ -66,8 +67,9 @@ public interface CdcConsumer {
     /**
      * Starts the consumer.
      * @param mreg Metric registry for consumer specific metrics.
+     * @param cdcDir Path to Change Data Capture Directory.
      */
-    public void start(MetricRegistry mreg);
+    public void start(MetricRegistry mreg, Path cdcDir);
 
     /**
      * Handles entry changes events.
@@ -131,7 +133,7 @@ public interface CdcConsumer {
 
     /**
      * Stops the consumer.
-     * This method can be invoked only after {@link #start(MetricRegistry)}.
+     * This method can be invoked only after {@link #start(MetricRegistry, Path)}.
      */
     public void stop();
 

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

@@ -339,7 +339,7 @@ public void runX() throws Exception {
                     committedSegmentOffset.value(walState.get1().fileOffset());
                 }
 
-                consumer.start(mreg, kctx.metric().registry(metricName("cdc", "consumer")));
+                consumer.start(mreg, kctx.metric().registry(metricName("cdc", "consumer")), cdcDir);
 
                 started = true;
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.cdc;
 
+import java.nio.file.Path;
 import java.util.EnumSet;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
@@ -188,8 +189,8 @@ public void onCacheDestroyEvents(Iterator<Integer> caches) {
      * @param cdcConsumerReg CDC consumer metric registry.
      * @throws IgniteCheckedException If failed.
      */
-    public void start(MetricRegistryImpl cdcReg, MetricRegistryImpl cdcConsumerReg) throws IgniteCheckedException {
-        consumer.start(cdcConsumerReg);
+    public void start(MetricRegistryImpl cdcReg, MetricRegistryImpl cdcConsumerReg, Path cdcDir) throws IgniteCheckedException {
+        consumer.start(cdcConsumerReg, cdcDir);
 
         evtsCnt = cdcReg.longMetric(EVTS_CNT, "Count of events processed by the consumer");
         lastEvtTs = cdcReg.longMetric(LAST_EVT_TIME, "Time of the last event process");
@@ -200,7 +201,7 @@ public void start(MetricRegistryImpl cdcReg, MetricRegistryImpl cdcConsumerReg)
 
     /**
      * Stops the consumer.
-     * This methods can be invoked only after {@link #start(MetricRegistryImpl, MetricRegistryImpl)}.
+     * This methods can be invoked only after {@link #start(MetricRegistryImpl, MetricRegistryImpl, Path)}.
      */
     public void stop() {
         consumer.stop();

modules/core/src/test/java/org/apache/ignite/cdc/AbstractCdcTest.java

@@ -308,7 +308,7 @@ public abstract static class TestCdcConsumer<T> implements CdcConsumer {
         private volatile boolean stopped;
 
         /** {@inheritDoc} */
-        @Override public void start(MetricRegistry mreg) {
+        @Override public void start(MetricRegistry mreg, Path cdcDir) {
             stopped = false;
         }
 
@@ -462,7 +462,7 @@ public static class TrackCacheEventsConsumer implements CdcConsumer {
         }
 
         /** {@inheritDoc} */
-        @Override public void start(MetricRegistry mreg) {
+        @Override public void start(MetricRegistry mreg, Path cdcDir) {
             // No-op.
         }
 

modules/core/src/test/java/org/apache/ignite/cdc/CdcNonDefaultWorkDirTest.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.cdc;
 
 import java.io.File;
+import java.nio.file.Path;
 import java.util.concurrent.CountDownLatch;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.configuration.DataStorageConfiguration;
@@ -92,8 +93,8 @@ public void testCdcStartWithNonDefaultWorkDir() throws Exception {
         CdcConfiguration cdcCfg = new CdcConfiguration();
 
         cdcCfg.setConsumer(new AbstractCdcTest.UserCdcConsumer() {
-            @Override public void start(MetricRegistry mreg) {
-                super.start(mreg);
+            @Override public void start(MetricRegistry mreg, Path cdcDir) {
+                super.start(mreg, cdcDir);
 
                 started.countDown();
             }

modules/core/src/test/java/org/apache/ignite/cdc/CdcPushMetricsExporterTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.cdc;
 
+import java.nio.file.Path;
 import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -201,10 +202,10 @@ public TestIgniteToIgniteConsumer(IgniteConfiguration destClusterCliCfg) {
         }
 
         /** {@inheritDoc} */
-        @Override public void start(MetricRegistry mreg) {
+        @Override public void start(MetricRegistry mreg, Path cdcDir) {
             Ignite ignite = Ignition.start(destClusterCliCfg);
 
-            super.start(mreg);
+            super.start(mreg, cdcDir);
 
             ignite.log().info("TestIgniteToIgniteConsumer started.");
         }

modules/core/src/test/java/org/apache/ignite/cdc/CdcSelfTest.java

@@ -19,6 +19,7 @@
 
 import java.io.File;
 import java.io.Serializable;
+import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -418,7 +419,7 @@ public void testReadOneByOneForBackup() throws Exception {
                     // No-op.
                 }
 
-                @Override public void start(MetricRegistry mreg) {
+                @Override public void start(MetricRegistry mreg, Path cdcDir) {
                     // No-op.
                 }
             };
@@ -512,7 +513,7 @@ public void testReadFromNextEntry() throws Exception {
                     // No-op.
                 }
 
-                @Override public void start(MetricRegistry mreg) {
+                @Override public void start(MetricRegistry mreg, Path cdcDir) {
                     // No-op.
                 }
             }, cfg));

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/cdc/CountingCdcConsumer.java

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.ducktest.tests.cdc;
 
+import java.nio.file.Path;
 import java.util.Iterator;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.ignite.IgniteLogger;
@@ -42,7 +43,7 @@ public class CountingCdcConsumer implements CdcConsumer {
     private final AtomicLong objectsConsumed = new AtomicLong();
 
     /** {@inheritDoc} */
-    @Override public void start(MetricRegistry mreg) {
+    @Override public void start(MetricRegistry mreg, Path cdcDir) {
         log.info("CountingCdcConsumer started");
     }