IGNITE-22530 CDC: Add regex filters for cache names

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

@@ -19,19 +19,19 @@
 
 import java.util.Iterator;
 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;
 import org.apache.ignite.internal.util.typedef.internal.A;
-import org.apache.ignite.internal.util.typedef.internal.CU;
 import org.apache.ignite.metric.MetricRegistry;
 import org.apache.ignite.resources.LoggerResource;
 
@@ -42,7 +42,7 @@
  *
  * @see AbstractCdcEventsApplier
  */
-public abstract class AbstractIgniteCdcStreamer implements CdcConsumer {
+public abstract class AbstractIgniteCdcStreamer implements CdcConsumerEx {
     /** */
     public static final String EVTS_SENT_CNT = "EventsCount";
 
@@ -73,8 +73,8 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumer {
     /** Cache names. */
     private Set<String> caches;
 
-    /** Cache IDs. */
-    protected Set<Integer> cachesIds;
+    /** Caches predicate. */
+    protected CachesPredicate cachesPredicate = new CachesPredicate();
 
     /** Maximum batch size. */
     protected int maxBatchSize;
@@ -100,12 +100,18 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumer {
 
     /** {@inheritDoc} */
     @Override public void start(MetricRegistry reg) {
+        //No-op
+    }
+
+    /** {@inheritDoc} */
+    @Override public void start(MetricRegistry reg, Iterator<CdcCacheEvent> cacheEvents) {
         A.notEmpty(caches, "caches");
 
-        cachesIds = caches.stream()
-            .mapToInt(CU::cacheId)
-            .boxed()
-            .collect(Collectors.toSet());
+        cachesPredicate.setLog(log);
+
+        cachesPredicate.setCaches(caches);
+
+        cacheEvents.forEachRemaining(evt -> cachesPredicate.onCacheEvent(evt.configuration().getName()));
 
         MetricRegistryImpl mreg = (MetricRegistryImpl)reg;
 
@@ -123,7 +129,7 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumer {
                 F.identity(),
                 true,
                 evt -> !onlyPrimary || evt.primary(),
-                evt -> F.isEmpty(cachesIds) || cachesIds.contains(evt.cacheId()),
+                evt -> cachesPredicate.test(evt.cacheId()),
                 evt -> evt.version().otherClusterVersion() == null));
 
             if (msgsSnt > 0) {
@@ -144,15 +150,13 @@ 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.
+            cachesPredicate.onCacheEvent(e.configuration().getName());
         });
     }
 
     /** {@inheritDoc} */
     @Override public void onCacheDestroy(Iterator<Integer> caches) {
-        caches.forEachRemaining(e -> {
-            // Just skip. Handle of cache events not supported.
-        });
+        caches.forEachRemaining(cachesPredicate::onCacheDestroy);
     }
 
     /** {@inheritDoc} */
@@ -238,6 +242,30 @@ public AbstractIgniteCdcStreamer setCaches(Set<String> caches) {
         return this;
     }
 
+    /**
+     * Sets include regex pattern for caches participating in CDC.
+     *
+     * @param includeRegex Include regex string
+     * @return {@code this} for chaining.
+     */
+    public AbstractIgniteCdcStreamer setIncludeCachesRegex(String includeRegex) {
+        cachesPredicate.setIncludeCacheTemplate(includeRegex);
+
+        return this;
+    }
+
+    /**
+     * Sets exclude regex pattern for caches participating in CDC.
+     *
+     * @param excludeRegex Exclude regex string
+     * @return {@code this} for chaining.
+     */
+    public AbstractIgniteCdcStreamer setExcludeCachesRegex(String excludeRegex) {
+        cachesPredicate.setExcludeCacheTemplate(excludeRegex);
+
+        return this;
+    }
+
     /**
      * Sets maximum batch size that will be applied to destination cluster.
      *

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

@@ -0,0 +1,152 @@
+/*
+ * 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.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.function.Predicate;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+import java.util.stream.Collectors;
+
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+
+/**
+ * Predicate for filtering {@link CdcEvent}s inside {@link CdcConsumer#onEvents(Iterator)}. Filters out events for
+ * following types of caches:
+ * <ol>
+ *   <li>Caches set in CDC configuration.</li>
+ *   <li>Caches that are added dynamically by user's cache regexp templates.</li>
+ * </ol>
+ */
+public class CachesPredicate implements Predicate<Integer> {
+    /** Include regex pattern for cache names. */
+    private Pattern includePtrn;
+
+    /** Exclude regex pattern for cache names. */
+    private Pattern excludePtrn;
+
+    /** Cache IDs. */
+    private Set<Integer> cacheIds;
+
+    /** Cache regex IDs. */
+    private final Set<Integer> cacheRegexIds = new ConcurrentSkipListSet<>();
+
+    /** Logger. */
+    private IgniteLogger log;
+
+    /**
+     * Sets cache ids of caches participating in CDC.
+     * @param caches Cache names.
+     */
+    public void setCaches(Collection<String> caches) {
+        cacheIds = caches.stream()
+            .mapToInt(CU::cacheId)
+            .boxed()
+            .collect(Collectors.toCollection(HashSet::new));
+    }
+
+    /**
+     * Sets include regex pattern for caches participating in CDC.
+     *
+     * @param includeRegex Include regex string
+     * @throws IgniteException If the template's syntax is invalid
+     */
+    public void setIncludeCacheTemplate(String includeRegex) {
+        try {
+            includePtrn = includeRegex != null ? Pattern.compile(includeRegex) : Pattern.compile("");
+        }
+        catch (PatternSyntaxException e) {
+            throw new IgniteException("Invalid cache regexp template", e);
+        }
+    }
+
+    /**
+     * Sets exclude regex pattern for caches participating in CDC.
+     *
+     * @param excludeRegex Exclude regex string
+     * @throws IgniteException If the template's syntax is invalid
+     */
+    public void setExcludeCacheTemplate(String excludeRegex) {
+        try {
+            excludePtrn = excludeRegex != null ? Pattern.compile(excludeRegex) : Pattern.compile("");
+        }
+        catch (PatternSyntaxException e) {
+            throw new IgniteException("Invalid cache regexp template", e);
+        }
+    }
+
+    /**
+     * @param log Logger.
+     */
+    public void setLog(IgniteLogger log) {
+        this.log = log;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean test(Integer cacheId) {
+        return cacheIds.contains(cacheId) || cacheRegexIds.contains(cacheId);
+    }
+
+    /**
+     * 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 onCacheEvent(String cacheName) {
+        if (excludePtrn.matcher(cacheName).matches())
+            return false;
+
+        if (includePtrn.matcher(cacheName).matches() && !cacheRegexIds.contains(CU.cacheId(cacheName))) {
+            cacheRegexIds.add(CU.cacheId(cacheName));
+
+            if (log.isInfoEnabled())
+                log.info("Cache [cacheName=" + cacheName + "] has been added to CDC");
+        }
+
+        return true;
+    }
+
+    /**
+     * Removes destroyed cache from replication.
+     * @param cacheId Cache id.
+     * */
+    public void onCacheDestroy(int cacheId) {
+        cacheRegexIds.remove(cacheId);
+
+        if (log.isInfoEnabled())
+            log.info("Cache [cacheId=" + cacheId + "] has been removed from CDC");
+    }
+
+    /**
+     * @return {@link Set} of cache ids participating in CDC.
+     */
+    public Set<Integer> getCacheIds() {
+        Set<Integer> cacheIds = new HashSet<>(this.cacheIds) ;
+
+        cacheIds.addAll(cacheRegexIds);
+
+        return cacheIds;
+    }
+}

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

@@ -17,6 +17,8 @@
 
 package org.apache.ignite.cdc;
 
+import java.util.Iterator;
+
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.Ignition;
 import org.apache.ignite.cdc.conflictresolve.CacheVersionConflictResolverImpl;
@@ -59,11 +61,11 @@ 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, Iterator<CdcCacheEvent> cacheEvents) {
+        super.start(mreg, cacheEvents);
 
         if (log.isInfoEnabled())
-            log.info("Ignite To Ignite Streamer [cacheIds=" + cachesIds + ']');
+            log.info("Ignite To Ignite Streamer [cacheIds=" + cachesPredicate.getCacheIds() + ']');
 
         A.notNull(destIgniteCfg, "Destination Ignite configuration.");
 

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

@@ -20,7 +20,9 @@
 import java.io.Serializable;
 import java.util.Set;
 import java.util.UUID;
+
 import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cdc.CachesPredicate;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.processors.cache.version.CacheVersionConflictResolver;
@@ -65,6 +67,9 @@ public class CacheVersionConflictResolverPluginProvider<C extends PluginConfigur
     /** Custom conflict resolver. */
     private CacheVersionConflictResolver resolver;
 
+    /** Caches predicate. */
+    protected CachesPredicate cachesPredicate = new CachesPredicate();
+
     /** Log. */
     private IgniteLogger log;
 
@@ -92,13 +97,14 @@ public CacheVersionConflictResolverPluginProvider() {
     @Override public void initExtensions(PluginContext ctx, ExtensionRegistry registry) {
         this.log = ctx.log(CacheVersionConflictResolverPluginProvider.class);
         this.provider = new CacheVersionConflictResolverCachePluginProvider<>(conflictResolveField, clusterId, resolver);
+        cachesPredicate.setLog(log);
     }
 
     /** {@inheritDoc} */
     @Override public CachePluginProvider createCacheProvider(CachePluginContext ctx) {
         String cacheName = ctx.igniteCacheConfiguration().getName();
 
-        if (caches.contains(cacheName)) {
+        if (caches.contains(cacheName) || cachesPredicate.onCacheEvent(cacheName)) {
             log.info("ConflictResolver provider set for cache [cacheName=" + cacheName + ']');
 
             return provider;
@@ -144,6 +150,16 @@ public void setConflictResolver(CacheVersionConflictResolver resolver) {
         this.resolver = resolver;
     }
 
+    /** @param includeRegex Include regex template */
+    public void setIncludeCachesRegex(String includeRegex) {
+        cachesPredicate.setIncludeCacheTemplate(includeRegex);
+    }
+
+    /** @param excludeRegex Exclude regex template */
+    public void setExcludeCachesRegex(String excludeRegex) {
+        cachesPredicate.setExcludeCacheTemplate(excludeRegex);
+    }
+
     /** {@inheritDoc} */
     @Override public void start(PluginContext ctx) {
         ((IgniteEx)ctx.grid()).context().cache().context().versions().dataCenterId(clusterId);