Added Dynamic Topic Mapping for the Source

Added a new interface `TopicMapper` with a `getTopic` method. Implementations can take the
Change Stream Document to determine the correct `topic` to publish the data to.

The Source connector includes a single implementation: `DefaultTopicMapper`.
The `DefaultTopicMapper` uses the 'ns' field to determine the topic to publish to. The following
configuration options can help configure topic mapping:

  * `topic.prefix` - Add a prefix to the topic eg: `<prefix>.<databaseName>.<collectionName>`
  * `topic.suffix` - Add a suffix to the topic eg: `<databaseName>.<collectionName>.<suffix>`
  * `topic.namespace.map` - Set a json map to maps change stream document namespaces to topics:
    For example: `{\"db\": \"dbTopic\", \"db.coll\": \"dbCollTopic\"}` will map all Change Stream
    Documents from the `db` database to `dbTopic.<collectionName>` apart from any documents
    from the `db.coll` namespace which map to the `dbCollTopic` topic.
  * `topic.mapper` - use an alternative mapping class, users can implement their own
    `TopicMapper` class to handle the mapping logic.

KAFKA-185

Author: mmolimar

CHANGELOG.md

@@ -8,6 +8,7 @@
   - [KAFKA-167](https://jira.mongodb.org/browse/KAFKA-167) Updated MongoDB Java Driver to 4.1.
   - [KAFKA-51](https://jira.mongodb.org/browse/KAFKA-51) Added sink support for MongoDB Changestream events.
   - [KAFKA-159](https://jira.mongodb.org/browse/KAFKA-159) Added dynamic namespace mapping for the sink connector.
+  - [KAFKA-185](https://jira.mongodb.org/browse/KAFKA-185) Added topic mapping for the source connector.
 
 ### Bug Fixes
   - [KAFKA-171](https://jira.mongodb.org/browse/KAFKA-171) Fixed bug which made the top level inferred schema optional

config/MongoSourceConnector.properties

@@ -8,6 +8,7 @@ database=test
 collection=source
 
 topic.prefix=
+topic.suffix=
 poll.max.batch.size=1000
 poll.await.time.ms=5000
 

src/integrationTest/java/com/mongodb/kafka/connect/source/MongoSourceTaskIntegrationTest.java

@@ -21,6 +21,7 @@
 import static com.mongodb.kafka.connect.mongodb.ChangeStreamOperations.createDropDatabase;
 import static com.mongodb.kafka.connect.mongodb.ChangeStreamOperations.createInserts;
 import static com.mongodb.kafka.connect.source.schema.SchemaUtils.assertStructsEquals;
+import static java.lang.String.format;
 import static java.util.Collections.singletonList;
 import static java.util.Collections.singletonMap;
 import static java.util.stream.Collectors.toList;
@@ -379,6 +380,49 @@ void testSourceCanHandleInvalidResumeTokenWhenErrorToleranceIsAll() {
     }
   }
 
+  @Test
+  @DisplayName("Ensure source sets the expected topic mapping")
+  void testSourceTopicMapping() {
+    assumeTrue(isGreaterThanThreeDotSix());
+    try (AutoCloseableSourceTask task = createSourceTask()) {
+
+      MongoDatabase db = getDatabaseWithPostfix();
+      MongoCollection<Document> coll1 = db.getCollection("coll1");
+      MongoCollection<Document> coll2 = db.getCollection("coll2");
+
+      insertMany(rangeClosed(1, 50), coll1, coll2);
+
+      HashMap<String, String> cfg =
+          new HashMap<String, String>() {
+            {
+              put(MongoSourceConfig.DATABASE_CONFIG, db.getName());
+              put(MongoSourceConfig.COPY_EXISTING_CONFIG, "true");
+              put(MongoSourceConfig.POLL_MAX_BATCH_SIZE_CONFIG, "100");
+              put(MongoSourceConfig.POLL_AWAIT_TIME_MS_CONFIG, "1000");
+              put(
+                  MongoSourceConfig.TOPIC_NAMESPACE_MAP_CONFIG,
+                  format(
+                      "{'%s': 'myDB', '%s': 'altDB.altColl'}",
+                      db.getName(), coll2.getNamespace().getFullName()));
+            }
+          };
+
+      task.start(cfg);
+
+      List<SourceRecord> firstPoll = getNextResults(task);
+      assertAll(
+          () -> assertSourceRecordValues(createInserts(1, 50), firstPoll, "myDB.coll1"),
+          () -> assertSourceRecordValues(createInserts(1, 50), firstPoll, "altDB.altColl"));
+
+      insertMany(rangeClosed(51, 100), coll1, coll2);
+
+      List<SourceRecord> secondPoll = getNextResults(task);
+      assertAll(
+          () -> assertSourceRecordValues(createInserts(51, 100), secondPoll, "myDB.coll1"),
+          () -> assertSourceRecordValues(createInserts(51, 100), secondPoll, "altDB.altColl"));
+    }
+  }
+
   @Test
   @DisplayName("Ensure source can use custom offset partition names")
   void testSourceCanUseCustomOffsetPartitionNames() {

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

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed 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 com.mongodb.kafka.connect.source;
+
+public interface Configurable {
+  default void configure(MongoSourceConfig configuration) {}
+}

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

@@ -25,6 +25,7 @@
 import static com.mongodb.kafka.connect.util.Validators.emptyString;
 import static com.mongodb.kafka.connect.util.Validators.errorCheckingValueValidator;
 import static java.lang.String.format;
+import static java.util.Arrays.asList;
 import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
 import static org.apache.kafka.common.config.ConfigDef.Width;
@@ -51,6 +52,7 @@
 
 import com.mongodb.kafka.connect.source.json.formatter.JsonWriterSettingsProvider;
 import com.mongodb.kafka.connect.source.schema.AvroSchema;
+import com.mongodb.kafka.connect.source.topic.mapping.TopicMapper;
 import com.mongodb.kafka.connect.util.ConfigHelper;
 import com.mongodb.kafka.connect.util.ConnectConfigException;
 import com.mongodb.kafka.connect.util.Validators;
@@ -126,13 +128,43 @@ public class MongoSourceConfig extends AbstractConfig {
   private static final String OUTPUT_SCHEMA_INFER_VALUE_DISPLAY =
       "Enable Infer Schemas for the value";
 
+  public static final String TOPIC_MAPPER_CONFIG = "topic.mapper";
+  private static final String TOPIC_MAPPER_DISPLAY = "The topic mapper class";
+  private static final String TOPIC_MAPPER_DOC =
+      "The class that determines the topic to write the source data to. "
+          + "By default this will be based on the 'ns' field in the change stream document, "
+          + "along with any configured prefix and suffix.";
+  private static final String TOPIC_MAPPER_DEFAULT =
+      "com.mongodb.kafka.connect.source.topic.mapping.DefaultTopicMapper";
+
   public static final String TOPIC_PREFIX_CONFIG = "topic.prefix";
   private static final String TOPIC_PREFIX_DOC =
       "Prefix to prepend to database & collection names to generate the name of the Kafka "
-          + "topic to publish data to.";
+          + "topic to publish data to. Used by the 'DefaultTopicMapper'.";
   private static final String TOPIC_PREFIX_DISPLAY = "Topic Prefix";
   private static final String TOPIC_PREFIX_DEFAULT = "";
 
+  public static final String TOPIC_SUFFIX_CONFIG = "topic.suffix";
+  private static final String TOPIC_SUFFIX_DOC =
+      "Suffix to append to database & collection names to generate the name of the Kafka "
+          + "topic to publish data to. Used by the 'DefaultTopicMapper'.";
+  private static final String TOPIC_SUFFIX_DISPLAY = "Topic Suffix";
+  private static final String TOPIC_SUFFIX_DEFAULT = "";
+
+  public static final String TOPIC_NAMESPACE_MAP_CONFIG = "topic.namespace.map";
+  private static final String TOPIC_NAMESPACE_MAP_DISPLAY = "The namespace to topic map";
+  private static final String TOPIC_NAMESPACE_MAP_DOC =
+      "A json map that maps change stream document namespaces to topics.\n"
+          + "For example: `{\"db\": \"dbTopic\", \"db.coll\": \"dbCollTopic\"}` will map all "
+          + "change stream documents from the `db` database to `dbTopic.<collectionName>` apart from"
+          + "any documents from the `db.coll` namespace which map to the `dbCollTopic` topic.\n"
+          + "If you want to map all messages to a single topic use `*`: "
+          + "For example: `{\"*\": \"everyThingTopic\", \"db.coll\": \"exceptionToTheRuleTopic\"}` "
+          + "will map all change stream documents to the `everyThingTopic` apart from the `db.coll` "
+          + "messages."
+          + "Note: Any prefix and suffix configuration will still apply.";
+  private static final String TOPIC_NAMESPACE_MAP_DEFAULT = "";
+
   public static final String PIPELINE_CONFIG = "pipeline";
   private static final String PIPELINE_DISPLAY = "The pipeline to apply to the change stream";
   private static final String PIPELINE_DOC =
@@ -292,7 +324,7 @@ public class MongoSourceConfig extends AbstractConfig {
 
   public static final ConfigDef CONFIG = createConfigDef();
   private static final List<Consumer<MongoSourceConfig>> INITIALIZERS =
-      singletonList(MongoSourceConfig::validateCollection);
+      asList(MongoSourceConfig::validateCollection, MongoSourceConfig::getTopicMapper);
 
   public enum OutputFormat {
     JSON,
@@ -314,6 +346,7 @@ public String value() {
   }
 
   private final ConnectionString connectionString;
+  private TopicMapper topicMapper;
 
   public MongoSourceConfig(final Map<?, ?> originals) {
     this(originals, true);
@@ -371,6 +404,16 @@ private void validateCollection() {
     }
   }
 
+  public TopicMapper getTopicMapper() {
+    if (topicMapper == null) {
+      topicMapper =
+          configureInstance(
+              createInstance(
+                  TOPIC_MAPPER_CONFIG, getString(TOPIC_MAPPER_CONFIG), TopicMapper.class));
+    }
+    return topicMapper;
+  }
+
   public JsonWriterSettings getJsonWriterSettings() {
     return createInstance(
             OUTPUT_JSON_FORMATTER_CONFIG,
@@ -388,6 +431,11 @@ public boolean tolerateErrors() {
         .equals(ErrorTolerance.ALL);
   }
 
+  private <T extends Configurable> T configureInstance(final T instance) {
+    instance.configure(this);
+    return instance;
+  }
+
   private static ConfigDef createConfigDef() {
     ConfigDef configDef =
         new ConfigDef() {
@@ -516,18 +564,6 @@ public Map<String, ConfigValue> validateAll(final Map<String, String> props) {
         Width.MEDIUM,
         COLLATION_DISPLAY);
 
-    configDef.define(
-        TOPIC_PREFIX_CONFIG,
-        Type.STRING,
-        TOPIC_PREFIX_DEFAULT,
-        null,
-        Importance.LOW,
-        TOPIC_PREFIX_DOC,
-        group,
-        ++orderInGroup,
-        Width.MEDIUM,
-        TOPIC_PREFIX_DISPLAY);
-
     configDef.define(
         POLL_MAX_BATCH_SIZE_CONFIG,
         Type.INT,
@@ -552,6 +588,54 @@ public Map<String, ConfigValue> validateAll(final Map<String, String> props) {
         Width.MEDIUM,
         POLL_AWAIT_TIME_MS_DISPLAY);
 
+    group = "Topic mapping";
+    orderInGroup = 0;
+    configDef.define(
+        TOPIC_MAPPER_CONFIG,
+        ConfigDef.Type.STRING,
+        TOPIC_MAPPER_DEFAULT,
+        Validators.matching(FULLY_QUALIFIED_CLASS_NAME),
+        ConfigDef.Importance.HIGH,
+        TOPIC_MAPPER_DOC,
+        group,
+        ++orderInGroup,
+        ConfigDef.Width.LONG,
+        TOPIC_MAPPER_DISPLAY);
+
+    configDef.define(
+        TOPIC_PREFIX_CONFIG,
+        Type.STRING,
+        TOPIC_PREFIX_DEFAULT,
+        null,
+        Importance.LOW,
+        TOPIC_PREFIX_DOC,
+        group,
+        ++orderInGroup,
+        Width.MEDIUM,
+        TOPIC_PREFIX_DISPLAY);
+    configDef.define(
+        TOPIC_SUFFIX_CONFIG,
+        Type.STRING,
+        TOPIC_SUFFIX_DEFAULT,
+        null,
+        Importance.LOW,
+        TOPIC_SUFFIX_DOC,
+        group,
+        ++orderInGroup,
+        Width.MEDIUM,
+        TOPIC_SUFFIX_DISPLAY);
+    configDef.define(
+        TOPIC_NAMESPACE_MAP_CONFIG,
+        Type.STRING,
+        TOPIC_NAMESPACE_MAP_DEFAULT,
+        errorCheckingValueValidator("A valid JSON document", ConfigHelper::documentFromString),
+        Importance.HIGH,
+        TOPIC_NAMESPACE_MAP_DOC,
+        group,
+        ++orderInGroup,
+        Width.MEDIUM,
+        TOPIC_NAMESPACE_MAP_DISPLAY);
+
     group = "Schema";
     orderInGroup = 0;
 

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

@@ -26,7 +26,6 @@
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.POLL_AWAIT_TIME_MS_CONFIG;
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.POLL_MAX_BATCH_SIZE_CONFIG;
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.PUBLISH_FULL_DOCUMENT_ONLY_CONFIG;
-import static com.mongodb.kafka.connect.source.MongoSourceConfig.TOPIC_PREFIX_CONFIG;
 import static com.mongodb.kafka.connect.source.heartbeat.HeartbeatManager.HEARTBEAT_KEY;
 import static com.mongodb.kafka.connect.source.producer.SchemaAndValueProducers.createKeySchemaAndValueProvider;
 import static com.mongodb.kafka.connect.source.producer.SchemaAndValueProducers.createValueSchemaAndValueProvider;
@@ -74,6 +73,7 @@
 import com.mongodb.kafka.connect.source.MongoSourceConfig.OutputFormat;
 import com.mongodb.kafka.connect.source.heartbeat.HeartbeatManager;
 import com.mongodb.kafka.connect.source.producer.SchemaAndValueProducer;
+import com.mongodb.kafka.connect.source.topic.mapping.TopicMapper;
 
 /**
  * A Kafka Connect source task that uses change streams to broadcast changes to the collection,
@@ -112,8 +112,6 @@ public final class MongoSourceTask extends SourceTask {
   private static final String CONNECTOR_TYPE = "source";
   public static final String ID_FIELD = "_id";
   private static final String COPY_KEY = "copy";
-  private static final String DB_KEY = "db";
-  private static final String COLL_KEY = "coll";
   private static final String NS_KEY = "ns";
   private static final String FULL_DOCUMENT = "fullDocument";
   private static final int NAMESPACE_NOT_FOUND_ERROR = 26;
@@ -187,10 +185,10 @@ public List<SourceRecord> poll() {
     final long startPoll = time.milliseconds();
     LOGGER.debug("Polling Start: {}", startPoll);
     List<SourceRecord> sourceRecords = new ArrayList<>();
+    TopicMapper topicMapper = sourceConfig.getTopicMapper();
     boolean publishFullDocumentOnly = sourceConfig.getBoolean(PUBLISH_FULL_DOCUMENT_ONLY_CONFIG);
     int maxBatchSize = sourceConfig.getInt(POLL_MAX_BATCH_SIZE_CONFIG);
     long nextUpdate = startPoll + sourceConfig.getLong(POLL_AWAIT_TIME_MS_CONFIG);
-    String prefix = sourceConfig.getString(TOPIC_PREFIX_CONFIG);
     Map<String, Object> partition = createPartitionMap(sourceConfig);
 
     SchemaAndValueProducer keySchemaAndValueProducer =
@@ -224,10 +222,7 @@ public List<SourceRecord> poll() {
           sourceOffset.put(COPY_KEY, "true");
         }
 
-        String topicName =
-            getTopicNameFromNamespace(
-                prefix, changeStreamDocument.getDocument("ns", new BsonDocument()));
-
+        String topicName = topicMapper.getTopic(changeStreamDocument);
         if (topicName.isEmpty()) {
           LOGGER.warn(
               "No topic set. Could not publish the message: {}", changeStreamDocument.toJson());
@@ -447,17 +442,6 @@ private boolean resumeTokenNotFound(final MongoCommandException e) {
             || errorMessage.contains(INVALID_RESUME_TOKEN));
   }
 
-  String getTopicNameFromNamespace(final String prefix, final BsonDocument namespaceDocument) {
-    String topicName = "";
-    if (namespaceDocument.containsKey(DB_KEY)) {
-      topicName = namespaceDocument.getString(DB_KEY).getValue();
-      if (namespaceDocument.containsKey(COLL_KEY)) {
-        topicName = format("%s.%s", topicName, namespaceDocument.getString(COLL_KEY).getValue());
-      }
-    }
-    return prefix.isEmpty() ? topicName : format("%s.%s", prefix, topicName);
-  }
-
   Map<String, Object> createPartitionMap(final MongoSourceConfig sourceConfig) {
     if (partitionMap == null) {
       String partitionName = sourceConfig.getString(MongoSourceConfig.OFFSET_PARTITION_NAME_CONFIG);