Added copy existing namespaces by regex configuration (#40)

* Added copy existing namespaces by regex configuration

KAFKA-147
Co-authored-by: abushmin <[email protected]>

Author: rozza

CHANGELOG.md

@@ -23,6 +23,8 @@
   - [KAFKA-155](https://jira.mongodb.org/browse/KAFKA-155) Fix business key update strategies to use dot notation for filters
   - [KAFKA-105](https://jira.mongodb.org/browse/KAFKA-105) Improve `errors.tolerance=all` support in the sink and source connectors.
   - [KAFKA-106](https://jira.mongodb.org/browse/KAFKA-106) Changed `max.num.retries` default to 1. A safer default especially as the driver now has retryable writes.
+  - [KAFKA-147](https://jira.mongodb.org/browse/KAFKA-147) Added `copy.existing.namespace.regex` configuration, that allows the filtering of namespaces to be copied.
+
 
 ## 1.2.0
   - [KAFKA-92](https://jira.mongodb.org/browse/KAFKA-92) Allow the Sink connector to use multiple tasks.

src/integrationTest/java/com/mongodb/kafka/connect/MongoSourceConnectorIntegrationTest.java

@@ -182,6 +182,45 @@ void testSourceLoadsDataFromCollectionCopyExistingBson() {
     assertProduced(createInserts(1, 100), coll, OutputFormat.BSON);
   }
 
+  @Test
+  @DisplayName("Ensure source loads data from collection with copy existing data by regex")
+  void testSourceLoadsDataFromCollectionCopyExistingByRegex() {
+    assumeTrue(isGreaterThanFourDotZero());
+    MongoDatabase db1 = getDatabaseWithPostfix();
+    MongoDatabase db2 = getDatabaseWithPostfix();
+    MongoDatabase db3 = getDatabaseWithPostfix();
+    MongoCollection<Document> coll1 = db1.getCollection("coll1");
+    MongoCollection<Document> coll21 = db2.getCollection("coll1");
+    MongoCollection<Document> coll22 = db2.getCollection("coll2");
+    MongoCollection<Document> coll23 = db2.getCollection("coll3");
+    MongoCollection<Document> coll3 = db3.getCollection("coll1");
+
+    insertMany(rangeClosed(1, 50), coll1);
+    insertMany(rangeClosed(1, 50), coll21);
+    insertMany(rangeClosed(1, 50), coll22);
+    insertMany(rangeClosed(1, 50), coll23);
+    insertMany(rangeClosed(1, 50), coll3);
+
+    Properties sourceProperties = new Properties();
+    sourceProperties.put(MongoSourceConfig.COPY_EXISTING_CONFIG, "true");
+    String namespaceRegex =
+        String.format("(%s\\.coll1|%s\\.coll(1|3))", db1.getName(), db2.getName());
+    sourceProperties.put(MongoSourceConfig.COPY_EXISTING_NAMESPACE_REGEX_CONFIG, namespaceRegex);
+
+    addSourceConnector(sourceProperties);
+
+    insertMany(rangeClosed(51, 100), coll1);
+    insertMany(rangeClosed(51, 100), coll21);
+    insertMany(rangeClosed(51, 100), coll22);
+    insertMany(rangeClosed(51, 100), coll23);
+    insertMany(rangeClosed(51, 100), coll3);
+    assertProduced(createInserts(1, 100), coll1);
+    assertProduced(createInserts(1, 100), coll21);
+    assertProduced(createInserts(51, 100), coll22);
+    assertProduced(createInserts(1, 100), coll23);
+    assertProduced(createInserts(51, 100), coll3);
+  }
+
   @Test
   @DisplayName("Ensure Schema Key and Value output")
   void testSchemaKeyAndValueOutput() {

src/main/java/com/mongodb/kafka/connect/source/MongoCopyDataManager.java

@@ -17,10 +17,12 @@
 
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.COLLECTION_CONFIG;
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.COPY_EXISTING_MAX_THREADS_CONFIG;
+import static com.mongodb.kafka.connect.source.MongoSourceConfig.COPY_EXISTING_NAMESPACE_REGEX_CONFIG;
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.COPY_EXISTING_PIPELINE_CONFIG;
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.COPY_EXISTING_QUEUE_SIZE_CONFIG;
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.DATABASE_CONFIG;
 import static java.util.Collections.singletonList;
+import static java.util.stream.Collectors.toList;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -31,7 +33,8 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
-import java.util.stream.Collectors;
+import java.util.function.Predicate;
+import java.util.regex.Pattern;
 
 import org.apache.kafka.connect.errors.ConnectException;
 import org.slf4j.Logger;
@@ -70,17 +73,8 @@ class MongoCopyDataManager implements AutoCloseable {
     this.sourceConfig = sourceConfig;
     this.mongoClient = mongoClient;
 
-    String database = sourceConfig.getString(DATABASE_CONFIG);
-    String collection = sourceConfig.getString(COLLECTION_CONFIG);
+    List<MongoNamespace> namespaces = selectNamespaces(sourceConfig, mongoClient);
 
-    List<MongoNamespace> namespaces;
-    if (database.isEmpty()) {
-      namespaces = getCollections(mongoClient);
-    } else if (collection.isEmpty()) {
-      namespaces = getCollections(mongoClient, database);
-    } else {
-      namespaces = singletonList(createNamespace(database, collection));
-    }
     LOGGER.info("Copying existing data on the following namespaces: {}", namespaces);
     namespacesToCopy = new AtomicInteger(namespaces.size());
     queue = new ArrayBlockingQueue<>(sourceConfig.getInt(COPY_EXISTING_QUEUE_SIZE_CONFIG));
@@ -142,6 +136,31 @@ private void putToQueue(final BsonDocument bsonDocument) {
     }
   }
 
+  static List<MongoNamespace> selectNamespaces(
+      final MongoSourceConfig sourceConfig, final MongoClient mongoClient) {
+
+    String database = sourceConfig.getString(DATABASE_CONFIG);
+    String collection = sourceConfig.getString(COLLECTION_CONFIG);
+    String namespacesRegex = sourceConfig.getString(COPY_EXISTING_NAMESPACE_REGEX_CONFIG);
+
+    List<MongoNamespace> namespaces;
+    if (database.isEmpty()) {
+      namespaces = getCollections(mongoClient);
+    } else if (collection.isEmpty()) {
+      namespaces = getCollections(mongoClient, database);
+    } else {
+      namespaces = singletonList(createNamespace(database, collection));
+    }
+
+    if (!namespacesRegex.isEmpty()) {
+      Predicate<String> predicate = Pattern.compile(namespacesRegex).asPredicate();
+      namespaces =
+          namespaces.stream().filter(n -> predicate.test(n.getFullName())).collect(toList());
+    }
+
+    return namespaces;
+  }
+
   static List<Bson> createPipeline(final MongoSourceConfig cfg, final MongoNamespace namespace) {
     List<Bson> pipeline = new ArrayList<>();
     cfg.getPipeline(COPY_EXISTING_PIPELINE_CONFIG).map(pipeline::addAll);
@@ -166,15 +185,15 @@ private static List<MongoNamespace> getCollections(final MongoClient mongoClient
         .filter(s -> !(s.startsWith("admin") || s.startsWith("config") || s.startsWith("local")))
         .map(d -> getCollections(mongoClient, d))
         .flatMap(Collection::stream)
-        .collect(Collectors.toList());
+        .collect(toList());
   }
 
   private static List<MongoNamespace> getCollections(
       final MongoClient mongoClient, final String database) {
     return mongoClient.getDatabase(database).listCollectionNames().into(new ArrayList<>()).stream()
         .filter(s -> !s.startsWith("system."))
         .map(c -> createNamespace(database, c))
-        .collect(Collectors.toList());
+        .collect(toList());
   }
 
   private static MongoNamespace createNamespace(final String database, final String collection) {

src/main/java/com/mongodb/kafka/connect/source/MongoSourceConfig.java

@@ -230,6 +230,16 @@ public class MongoSourceConfig extends AbstractConfig {
           + "Example: `[{\"$match\": {\"closed\": \"false\"}}]`";
   private static final String COPY_EXISTING_PIPELINE_DEFAULT = "";
 
+  public static final String COPY_EXISTING_NAMESPACE_REGEX_CONFIG = "copy.existing.namespace.regex";
+  private static final String COPY_EXISTING_NAMESPACE_REGEX_DISPLAY =
+      "Copy existing namespace regex";
+  private static final String COPY_EXISTING_NAMESPACE_REGEX_DOC =
+      "Use a regular expression to define from which existing namespaces data should be copied from."
+          + " A namespace is the database name and collection separated by a period e.g. `database.collection`.\n"
+          + " Example: The following regular expression will only include collections starting with `a` "
+          + "in the `demo` database: `demo\\.a.*`";
+  private static final String COPY_EXISTING_NAMESPACE_REGEX_DEFAULT = "";
+
   public static final String ERRORS_TOLERANCE_CONFIG = "errors.tolerance";
   public static final String ERRORS_TOLERANCE_DISPLAY = "Error Tolerance";
   public static final ErrorTolerance ERRORS_TOLERANCE_DEFAULT = ErrorTolerance.NONE;
@@ -659,6 +669,18 @@ public Map<String, ConfigValue> validateAll(final Map<String, String> props) {
         Width.MEDIUM,
         COPY_EXISTING_PIPELINE_DISPLAY);
 
+    configDef.define(
+        COPY_EXISTING_NAMESPACE_REGEX_CONFIG,
+        Type.STRING,
+        COPY_EXISTING_NAMESPACE_REGEX_DEFAULT,
+        Validators.emptyString().or(Validators.isAValidRegex()),
+        Importance.MEDIUM,
+        COPY_EXISTING_NAMESPACE_REGEX_DOC,
+        group,
+        ++orderInGroup,
+        Width.MEDIUM,
+        COPY_EXISTING_NAMESPACE_REGEX_DISPLAY);
+
     group = "Errors";
     orderInGroup = 0;