IGNITE-22530 CDC: Add regex filters for cache names

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

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cdc;
+
+import java.nio.file.Path;
+
+import org.apache.ignite.cdc.CdcConsumer;
+import org.apache.ignite.metric.MetricRegistry;
+
+/**
+ * Extended CdcConsumer interface which provides overloaded {@link CdcConsumerEx#start(MetricRegistry, Path)} 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.
+     */
+    void start(MetricRegistry mreg, Path cdcDir);
+}

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")), ft.walCdc().toPath());
 
                 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;
@@ -186,10 +187,14 @@ 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.
      * @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 {
+        if (consumer instanceof CdcConsumerEx)
+            ((CdcConsumerEx) consumer).start(cdcConsumerReg, cdcDir);
+        else
+            consumer.start(cdcConsumerReg);
 
         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 +205,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();