IGNITE-22530 Add CdcConsumerEx interface

Author: Andrei Nadyktov

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

@@ -19,7 +19,6 @@
 
 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;
@@ -195,7 +194,7 @@ public void testDeleteLostSegmentLinksApplicationNotClosed() throws Exception {
         CdcConfiguration cfg = new CdcConfiguration();
 
         cfg.setConsumer(new UserCdcConsumer() {
-            @Override public void start(MetricRegistry mreg, Path cdcDir) {
+            @Override public void start(MetricRegistry mreg) {
                 appStarted.countDown();
             }
         });

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

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

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

@@ -17,7 +17,6 @@
 
 package org.apache.ignite.cdc;
 
-import java.nio.file.Path;
 import java.util.Iterator;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteBinary;
@@ -34,7 +33,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, Path)}.</li>
+ *     <li>Start of the consumer {@link #start(MetricRegistry)}.</li>
  *     <li>Notification of the consumer by the {@link #onEvents(Iterator)} call.</li>
  *     <li>Stop of the consumer {@link #stop()}.</li>
  * </ul>
@@ -67,9 +66,8 @@ 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, Path cdcDir);
+    public void start(MetricRegistry mreg);
 
     /**
      * Handles entry changes events.
@@ -133,7 +131,7 @@ public interface CdcConsumer {
 
     /**
      * Stops the consumer.
-     * This method can be invoked only after {@link #start(MetricRegistry, Path)}.
+     * This method can be invoked only after {@link #start(MetricRegistry)}.
      */
     public void stop();
 

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

@@ -0,0 +1,34 @@
+/*
+ * 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.cdc;
+
+import java.nio.file.Path;
+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")), cdcDir);
+                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

@@ -27,6 +27,7 @@
 import org.apache.ignite.binary.BinaryType;
 import org.apache.ignite.cdc.CdcCacheEvent;
 import org.apache.ignite.cdc.CdcConsumer;
+import org.apache.ignite.cdc.CdcConsumerEx;
 import org.apache.ignite.cdc.CdcEvent;
 import org.apache.ignite.cdc.TypeMapping;
 import org.apache.ignite.internal.pagemem.wal.WALIterator;
@@ -187,10 +188,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, Path cdcDir) throws IgniteCheckedException {
-        consumer.start(cdcConsumerReg, cdcDir);
+        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");

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, Path cdcDir) {
+        @Override public void start(MetricRegistry mreg) {
             stopped = false;
         }
 
@@ -462,7 +462,7 @@ public static class TrackCacheEventsConsumer implements CdcConsumer {
         }
 
         /** {@inheritDoc} */
-        @Override public void start(MetricRegistry mreg, Path cdcDir) {
+        @Override public void start(MetricRegistry mreg) {
             // No-op.
         }
 

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

@@ -18,7 +18,6 @@
 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;
@@ -93,8 +92,8 @@ public void testCdcStartWithNonDefaultWorkDir() throws Exception {
         CdcConfiguration cdcCfg = new CdcConfiguration();
 
         cdcCfg.setConsumer(new AbstractCdcTest.UserCdcConsumer() {
-            @Override public void start(MetricRegistry mreg, Path cdcDir) {
-                super.start(mreg, cdcDir);
+            @Override public void start(MetricRegistry mreg) {
+                super.start(mreg);
 
                 started.countDown();
             }

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

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

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

@@ -19,7 +19,6 @@
 
 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;
@@ -419,7 +418,7 @@ public void testReadOneByOneForBackup() throws Exception {
                     // No-op.
                 }
 
-                @Override public void start(MetricRegistry mreg, Path cdcDir) {
+                @Override public void start(MetricRegistry mreg) {
                     // No-op.
                 }
             };
@@ -513,7 +512,7 @@ public void testReadFromNextEntry() throws Exception {
                     // No-op.
                 }
 
-                @Override public void start(MetricRegistry mreg, Path cdcDir) {
+                @Override public void start(MetricRegistry mreg) {
                     // No-op.
                 }
             }, cfg));