IGNITE-22530 CDC: Add regex filters for cache names

modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractIgniteCdcStreamer.java

@@ -17,16 +17,20 @@
 
 package org.apache.ignite.cdc;
 
+import java.nio.file.Path;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
+
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.binary.BinaryType;
 import org.apache.ignite.internal.binary.BinaryContext;
 import org.apache.ignite.internal.binary.BinaryMetadata;
 import org.apache.ignite.internal.binary.BinaryTypeImpl;
+import org.apache.ignite.internal.cdc.CdcConsumerEx;
 import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
 import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric;
 import org.apache.ignite.internal.util.typedef.F;
@@ -42,7 +46,7 @@
  *
  * @see AbstractCdcEventsApplier
  */
-public abstract class AbstractIgniteCdcStreamer implements CdcConsumer {
+public abstract class AbstractIgniteCdcStreamer implements CdcConsumerEx {
     /** */
     public static final String EVTS_SENT_CNT = "EventsCount";
 
@@ -67,12 +71,24 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumer {
     /** */
     public static final String LAST_EVT_SENT_TIME_DESC = "Timestamp of last applied event to destination cluster";
 
+    /** */
+    public static final String DFLT_REGEXP = "";
+
     /** Handle only primary entry flag. */
     private boolean onlyPrimary = DFLT_IS_ONLY_PRIMARY;
 
     /** Cache names. */
     private Set<String> caches;
 
+    /** Regexp manager. */
+    private CdcRegexManager regexManager;
+
+    /** Include regex template for cache names. */
+    private String includeTemplate = DFLT_REGEXP;
+
+    /** Exclude regex template for cache names. */
+    private String excludeTemplate = DFLT_REGEXP;
+
     /** Cache IDs. */
     protected Set<Integer> cachesIds;
 
@@ -100,13 +116,27 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumer {
 
     /** {@inheritDoc} */
     @Override public void start(MetricRegistry reg) {
+        //No-op
+    }
+
+    /** {@inheritDoc} */
+    @Override public void start(MetricRegistry reg, Path cdcDir, List<String> cacheNames) {
         A.notEmpty(caches, "caches");
 
+        regexManager = new CdcRegexManager();
+
         cachesIds = caches.stream()
             .mapToInt(CU::cacheId)
             .boxed()
             .collect(Collectors.toSet());
 
+        regexManager.compileRegexp(includeTemplate, excludeTemplate);
+
+        cacheNames.stream()
+            .filter(regexManager::matchesFilters)
+            .map(CU::cacheId)
+            .forEach(cachesIds::add);
+
         MetricRegistryImpl mreg = (MetricRegistryImpl)reg;
 
         this.evtsCnt = mreg.longMetric(EVTS_SENT_CNT, EVTS_SENT_CNT_DESC);
@@ -144,15 +174,26 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumer {
     /** {@inheritDoc} */
     @Override public void onCacheChange(Iterator<CdcCacheEvent> cacheEvents) {
         cacheEvents.forEachRemaining(e -> {
-            // Just skip. Handle of cache events not supported.
+            matchWithRegex(e.configuration().getName());
         });
     }
 
+    /**
+     * Finds match between cache name and user's regex templates.
+     * If match is found, adds this cache's id to id's list.
+     *
+     * @param cacheName Cache name.
+     */
+    private void matchWithRegex(String cacheName) {
+        int cacheId = CU.cacheId(cacheName);
+
+        if (!cachesIds.contains(cacheId) && regexManager.matchesFilters(cacheName))
+            cachesIds.add(cacheId);
+    }
+
     /** {@inheritDoc} */
     @Override public void onCacheDestroy(Iterator<Integer> caches) {
-        caches.forEachRemaining(e -> {
-            // Just skip. Handle of cache events not supported.
-        });
+        caches.forEachRemaining(cachesIds::remove);
     }
 
     /** {@inheritDoc} */
@@ -238,6 +279,30 @@ public AbstractIgniteCdcStreamer setCaches(Set<String> caches) {
         return this;
     }
 
+    /**
+     * Sets include regex pattern that participates in CDC.
+     *
+     * @param includeTemplate Include regex template
+     * @return {@code this} for chaining.
+     */
+    public AbstractIgniteCdcStreamer setIncludeTemplate(String includeTemplate) {
+        this.includeTemplate = includeTemplate;
+
+        return this;
+    }
+
+    /**
+     * Sets exclude regex pattern that participates in CDC.
+     *
+     * @param excludeTemplate Exclude regex template
+     * @return {@code this} for chaining.
+     */
+    public AbstractIgniteCdcStreamer setExcludeTemplate(String excludeTemplate) {
+        this.excludeTemplate = excludeTemplate;
+
+        return this;
+    }
+
     /**
      * Sets maximum batch size that will be applied to destination cluster.
      *

modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcRegexManager.java

@@ -0,0 +1,63 @@
+/*
+ * 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.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+import org.apache.ignite.IgniteException;
+
+/**
+ * Contains logic to process user's regexp patterns for CDC.
+ */
+public class CdcRegexManager {
+
+    /** Include regex pattern for cache names. */
+    private Pattern includeFilter;
+
+    /** Exclude regex pattern for cache names. */
+    private Pattern excludeFilter;
+
+    /**
+     * Matches cache name with compiled regex patterns.
+     *
+     * @param cacheName Cache name.
+     * @return True if cache name matches include pattern and doesn't match exclude pattern.
+     */
+    public boolean matchesFilters(String cacheName) {
+        return includeFilter.matcher(cacheName).matches() && !excludeFilter.matcher(cacheName).matches();
+    }
+
+    /**
+     * Compiles regex patterns from user templates.
+     *
+     * @param includeTemplate Include regex template.
+     * @param excludeTemplate Exclude regex template.
+     * @throws IgniteException If the template's syntax is invalid
+     */
+    public void compileRegexp(String includeTemplate, String excludeTemplate) {
+        try {
+            includeFilter = Pattern.compile(includeTemplate);
+
+            excludeFilter = Pattern.compile(excludeTemplate);
+        }
+        catch (PatternSyntaxException e) {
+            throw new IgniteException("Invalid cache regexp template.", e);
+        }
+    }
+}

modules/cdc-ext/src/main/java/org/apache/ignite/cdc/IgniteToIgniteCdcStreamer.java

@@ -17,6 +17,9 @@
 
 package org.apache.ignite.cdc;
 
+import java.nio.file.Path;
+import java.util.List;
+
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.Ignition;
 import org.apache.ignite.cdc.conflictresolve.CacheVersionConflictResolverImpl;
@@ -59,8 +62,8 @@ public class IgniteToIgniteCdcStreamer extends AbstractIgniteCdcStreamer impleme
     private volatile boolean alive = true;
 
     /** {@inheritDoc} */
-    @Override public void start(MetricRegistry mreg) {
-        super.start(mreg);
+    @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/conflictresolve/CacheVersionConflictResolverPluginProvider.java

@@ -20,6 +20,10 @@
 import java.io.Serializable;
 import java.util.Set;
 import java.util.UUID;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.internal.IgniteEx;
@@ -41,6 +45,9 @@
  * @see CacheVersionConflictResolver
  */
 public class CacheVersionConflictResolverPluginProvider<C extends PluginConfiguration> implements PluginProvider<C> {
+    /** */
+    public static final String DFLT_REGEXP = "";
+
     /** Cluster id. */
     private byte clusterId;
 
@@ -65,6 +72,12 @@ public class CacheVersionConflictResolverPluginProvider<C extends PluginConfigur
     /** Custom conflict resolver. */
     private CacheVersionConflictResolver resolver;
 
+    /** Include regex templates for cache names. */
+    private String includeTemplate = DFLT_REGEXP;
+
+    /** Exclude regex templates for cache names. */
+    private String excludeTemplate = DFLT_REGEXP;
+
     /** Log. */
     private IgniteLogger log;
 
@@ -98,7 +111,7 @@ public CacheVersionConflictResolverPluginProvider() {
     @Override public CachePluginProvider createCacheProvider(CachePluginContext ctx) {
         String cacheName = ctx.igniteCacheConfiguration().getName();
 
-        if (caches.contains(cacheName)) {
+        if (caches.contains(cacheName) || matchesFilters(cacheName)) {
             log.info("ConflictResolver provider set for cache [cacheName=" + cacheName + ']');
 
             return provider;
@@ -144,6 +157,16 @@ public void setConflictResolver(CacheVersionConflictResolver resolver) {
         this.resolver = resolver;
     }
 
+    /** @param includeTemplate Include regex template */
+    public void setIncludeTemplate(String includeTemplate) {
+        this.includeTemplate = includeTemplate;
+    }
+
+    /** @param excludeTemplate Exclude regex template */
+    public void setExcludeTemplate(String excludeTemplate) {
+        this.excludeTemplate = excludeTemplate;
+    }
+
     /** {@inheritDoc} */
     @Override public void start(PluginContext ctx) {
         ((IgniteEx)ctx.grid()).context().cache().context().versions().dataCenterId(clusterId);
@@ -178,4 +201,24 @@ public void setConflictResolver(CacheVersionConflictResolver resolver) {
     @Nullable @Override public <T> T createComponent(PluginContext ctx, Class<T> cls) {
         return null;
     }
+
+    /**
+     * Matches cache name with compiled regex patterns.
+     *
+     * @param cacheName Cache name.
+     * @return True if cache name matches include pattern and doesn't match exclude pattern.
+     * @throws IgniteException If the template's syntax is invalid
+     */
+    private boolean matchesFilters(String cacheName) {
+        try {
+            boolean matchesInclude = Pattern.compile(includeTemplate).matcher(cacheName).matches();
+
+            boolean matchesExclude = Pattern.compile(excludeTemplate).matcher(cacheName).matches();
+
+            return matchesInclude && !matchesExclude;
+        }
+        catch (PatternSyntaxException e) {
+            throw new IgniteException("Invalid cache regexp template.", e);
+        }
+    }
 }

modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/AbstractKafkaToIgniteCdcStreamer.java

@@ -19,11 +19,11 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.cdc.AbstractCdcEventsApplier;
@@ -152,13 +152,13 @@ public AbstractKafkaToIgniteCdcStreamer(Properties kafkaProps, KafkaToIgniteCdcS
     protected void runAppliers() {
         AtomicBoolean stopped = new AtomicBoolean();
 
-        Set<Integer> caches = null;
+        Set<Integer> caches = new HashSet<>();
 
         if (!F.isEmpty(streamerCfg.getCaches())) {
             checkCaches(streamerCfg.getCaches());
 
-            caches = streamerCfg.getCaches().stream()
-                .map(CU::cacheId).collect(Collectors.toSet());
+            streamerCfg.getCaches().stream()
+                .map(CU::cacheId).forEach(caches::add);
         }
 
         KafkaToIgniteMetadataUpdater metaUpdr = new KafkaToIgniteMetadataUpdater(
@@ -181,7 +181,8 @@ protected void runAppliers() {
                 caches,
                 metaUpdr,
                 stopped,
-                metrics
+                metrics,
+                this
             );
 
             addAndStart("applier-thread-" + cntr++, applier);
@@ -252,6 +253,13 @@ private <T extends AutoCloseable & Runnable> void addAndStart(String threadName,
     /** Checks that configured caches exist in a destination cluster. */
     protected abstract void checkCaches(Collection<String> caches);
 
+    /**
+     * Get cache names from client.
+     *
+     * @return Cache names.
+     * */
+    protected abstract Collection<String> getCaches();
+
     /** */
     private void ackAsciiLogo(IgniteLogger log) {
         String ver = "ver. " + ACK_VER_STR;