IGNITE-22530 Refactor code

Author: Andrei Nadyktov

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

@@ -70,9 +70,6 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumerEx {
     /** */
     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;
 
@@ -83,10 +80,10 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumerEx {
     private CdcRegexManager regexManager;
 
     /** Include regex template for cache names. */
-    private String includeTemplate = DFLT_REGEXP;
+    private String includeTemplate;
 
     /** Exclude regex template for cache names. */
-    private String excludeTemplate = DFLT_REGEXP;
+    private String excludeTemplate;
 
     /** Cache IDs. */
     protected Set<Integer> cachesIds;
@@ -122,7 +119,7 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumerEx {
     @Override public void start(MetricRegistry reg, List<String> cacheNames) {
         A.notEmpty(caches, "caches");
 
-        regexManager = new CdcRegexManager();
+        regexManager = new CdcRegexManager(log);
 
         cachesIds = caches.stream()
             .mapToInt(CU::cacheId)

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

@@ -21,6 +21,7 @@
 import java.util.regex.PatternSyntaxException;
 
 import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
 
 /**
  * Contains logic to process user's regexp patterns for CDC.
@@ -33,13 +34,31 @@ public class CdcRegexManager {
     /** Exclude regex pattern for cache names. */
     private Pattern excludeFilter;
 
+    /** Logger. */
+    private IgniteLogger log;
+
+    /**
+     *
+     * @param log Logger.
+     */
+    public CdcRegexManager(IgniteLogger log) {
+        this.log = log;
+    }
+
     /**
      * 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) {
+        if (includeFilter.matcher(cacheName).matches() && excludeFilter.matcher(cacheName).matches()) {
+            log.warning("Cache name matches both include and exclude regexp templates. Will ignore to prevent " +
+                "ambiguity [cacheName=" + cacheName + ", includeTemplate=" + includeFilter + ", excludeTemplate=" +
+                excludeFilter + "]");
+
+            return false;
+        }
         return includeFilter.matcher(cacheName).matches() && !excludeFilter.matcher(cacheName).matches();
     }
 
@@ -52,9 +71,9 @@ public boolean matchesFilters(String cacheName) {
      */
     public void compileRegexp(String includeTemplate, String excludeTemplate) {
         try {
-            includeFilter = Pattern.compile(includeTemplate);
+            includeFilter = includeTemplate != null ? Pattern.compile(includeTemplate) : Pattern.compile("");
 
-            excludeFilter = Pattern.compile(excludeTemplate);
+            excludeFilter = excludeTemplate != null ? Pattern.compile(excludeTemplate) : Pattern.compile("");
         }
         catch (PatternSyntaxException e) {
             throw new IgniteException("Invalid cache regexp template", e);

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

@@ -115,9 +115,6 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumerEx {
     /** Count of metadata markers sent description. */
     public static final String MARKERS_SENT_CNT_DESC = "Count of metadata markers sent to Kafka";
 
-    /** */
-    public static final String DFLT_REGEXP = "";
-
     /** Default value for the flag that indicates whether entries only from primary nodes should be handled. */
     public static final boolean DFLT_IS_ONLY_PRIMARY = false;
 
@@ -156,10 +153,10 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumerEx {
     private CdcRegexManager regexManager;
 
     /** Include regex template for cache names. */
-    private String includeTemplate = DFLT_REGEXP;
+    private String includeTemplate;
 
     /** Exclude regex template for cache names. */
-    private String excludeTemplate = DFLT_REGEXP;
+    private String excludeTemplate;
 
     /** Max batch size. */
     private int maxBatchSz = DFLT_MAX_BATCH_SIZE;
@@ -346,7 +343,7 @@ private <T> void sendOneBatch(
         kafkaProps.setProperty(KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
         kafkaProps.setProperty(VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
 
-        regexManager = new CdcRegexManager();
+        regexManager = new CdcRegexManager(log);
 
         cachesIds = caches.stream()
             .map(CU::cacheId)