Fix catalog.include option

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## [Unreleased]
 
+-   Fix `catalog.include` option and add more logging.
+
 ## [0.1.1] - 2024-01-26
 
 -   Add `catalog.include` and `catalog.exclude` options.

src/main/java/com/getindata/flink/connector/jdbc/catalog/ElasticJdbcCatalogFactoryOptions.java

@@ -85,7 +85,7 @@ public class ElasticJdbcCatalogFactoryOptions {
             .withDescription("Comma-separated list of index/datastream exclude patterns.");
 
     public static final ConfigOption<String> INCLUDE =
-        ConfigOptions.key("catalog.exclude")
+        ConfigOptions.key("catalog.include")
             .stringType()
             .noDefaultValue()
             .withDescription("Comma-separated list of index/datastream include patterns.");

src/main/java/com/getindata/flink/connector/jdbc/catalog/IndexFilterResolver.java

@@ -1,6 +1,8 @@
 package com.getindata.flink.connector.jdbc.catalog;
 
 import org.apache.flink.shaded.guava30.com.google.common.base.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
 import java.util.List;
@@ -11,6 +13,8 @@
 
 class IndexFilterResolver {
 
+    private static final Logger LOG = LoggerFactory.getLogger(IndexFilterResolver.class);
+
     static IndexFilterResolver acceptAll() {
         return new IndexFilterResolver(emptyList(), emptyList());
     }
@@ -40,12 +44,17 @@ private static List<Pattern> parseRaw(String commaSeparatedList) {
     boolean isAccepted(String objectName) {
         if (!includePatterns.isEmpty()) {
             if (includePatterns.stream().noneMatch(pattern -> pattern.matcher(objectName).matches())) {
+                LOG.info("'{}' does not match any include pattern. Include patterns='{}'.", objectName, includePatterns);
                 return false;
             }
         }
         if (!excludePatterns.isEmpty()) {
-            return excludePatterns.stream().noneMatch(pattern -> pattern.matcher(objectName).matches());
+            if (excludePatterns.stream().anyMatch(pattern -> pattern.matcher(objectName).matches())) {
+                LOG.info("'{}' matches exclude pattern; exclude patterns='{}'.", objectName, excludePatterns);
+                return false;
+            }
         }
+        LOG.info("'{}' matches include pattern and does not match any exclude pattern.", objectName);
         return true;
     }