Added Dynamic Namespace Mapping for the Sink

Added a new interface `NamespaceMapper` with a `getNamespace` method.
Implementations can use either the raw `SinkRecord` or the `SinkDocument` to
determine the correct `MongoNamespace` to sink the data to.

The sink connector includes two implementations:

 * `DefaultNamespaceMapper`
   Uses the configured database and the collection or topic name if no collection configured as the namespace.
 * `FieldPathNamespaceMapper`
   Uses a string from either the key or value document as the database or collection name.

KAFKA-159

Author: rozza

CHANGELOG.md

@@ -7,6 +7,7 @@
 ### Improvements
   - [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.
 
 ### Bug Fixes
   - [KAFKA-171](https://jira.mongodb.org/browse/KAFKA-171) Fixed bug which made the top level inferred schema optional

src/main/java/com/mongodb/kafka/connect/sink/MongoProcessedSinkRecordData.java

@@ -0,0 +1,118 @@
+/*
+ * 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.sink;
+
+import java.util.Optional;
+import java.util.function.Supplier;
+
+import org.apache.kafka.connect.errors.DataException;
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.bson.BsonDocument;
+
+import com.mongodb.MongoNamespace;
+import com.mongodb.client.model.WriteModel;
+
+import com.mongodb.kafka.connect.sink.converter.SinkConverter;
+import com.mongodb.kafka.connect.sink.converter.SinkDocument;
+import com.mongodb.kafka.connect.sink.writemodel.strategy.WriteModelStrategyHelper;
+
+final class MongoProcessedSinkRecordData {
+  private static final Logger LOGGER = LoggerFactory.getLogger(MongoProcessedSinkRecordData.class);
+  private static final SinkConverter SINK_CONVERTER = new SinkConverter();
+
+  private final MongoSinkTopicConfig config;
+  private final MongoNamespace namespace;
+  private final SinkRecord sinkRecord;
+  private final SinkDocument sinkDocument;
+  private final WriteModel<BsonDocument> writeModel;
+
+  MongoProcessedSinkRecordData(final SinkRecord sinkRecord, final MongoSinkConfig sinkConfig) {
+    this.sinkRecord = sinkRecord;
+    this.config = sinkConfig.getMongoSinkTopicConfig(sinkRecord.topic());
+    this.sinkDocument = SINK_CONVERTER.convert(sinkRecord);
+    this.namespace = createNamespace();
+    this.writeModel = createWriteModel();
+  }
+
+  public MongoSinkTopicConfig getConfig() {
+    return config;
+  }
+
+  public MongoNamespace getNamespace() {
+    return namespace;
+  }
+
+  public SinkRecord getSinkRecord() {
+    return sinkRecord;
+  }
+
+  public WriteModel<BsonDocument> getWriteModel() {
+    if (writeModel == null) {
+      throw new DataException("Unable to create a valid WriteModel for the SinkRecord");
+    }
+    return writeModel;
+  }
+
+  public boolean canProcess() {
+    return namespace != null && writeModel != null;
+  }
+
+  private MongoNamespace createNamespace() {
+    return tryProcess(
+            () -> Optional.of(config.getNamespaceMapper().getNamespace(sinkRecord, sinkDocument)))
+        .orElse(null);
+  }
+
+  private WriteModel<BsonDocument> createWriteModel() {
+    return config.getCdcHandler().isPresent() ? buildWriteModelCDC() : buildWriteModel();
+  }
+
+  private WriteModel<BsonDocument> buildWriteModel() {
+    return tryProcess(
+            () -> {
+              config
+                  .getPostProcessors()
+                  .getPostProcessorList()
+                  .forEach(pp -> pp.process(sinkDocument, sinkRecord));
+              return WriteModelStrategyHelper.createWriteModel(config, sinkDocument);
+            })
+        .orElse(null);
+  }
+
+  private WriteModel<BsonDocument> buildWriteModelCDC() {
+    return tryProcess(
+            () -> config.getCdcHandler().flatMap(cdcHandler -> cdcHandler.handle(sinkDocument)))
+        .orElse(null);
+  }
+
+  private <T> Optional<T> tryProcess(final Supplier<Optional<T>> supplier) {
+    try {
+      return supplier.get();
+    } catch (Exception e) {
+      if (config.logErrors()) {
+        LOGGER.error("Unable to process record {}", sinkRecord, e);
+      }
+      if (!config.tolerateErrors()) {
+        throw e;
+      }
+    }
+    return Optional.empty();
+  }
+}

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkRecordProcessor.java

@@ -0,0 +1,71 @@
+/*
+ * 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.sink;
+
+import static com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.MAX_BATCH_SIZE_CONFIG;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class MongoSinkRecordProcessor {
+  private static final Logger LOGGER = LoggerFactory.getLogger(MongoSinkRecordProcessor.class);
+
+  static List<List<MongoProcessedSinkRecordData>> orderedGroupByTopicAndNamespace(
+      final Collection<SinkRecord> records, final MongoSinkConfig sinkConfig) {
+    LOGGER.debug("Number of sink records to process: {}", records.size());
+
+    List<List<MongoProcessedSinkRecordData>> orderedProcessedSinkRecordData = new ArrayList<>();
+    List<MongoProcessedSinkRecordData> currentGroup = new ArrayList<>();
+    MongoProcessedSinkRecordData previous = null;
+
+    for (SinkRecord record : records) {
+      MongoProcessedSinkRecordData processedData =
+          new MongoProcessedSinkRecordData(record, sinkConfig);
+
+      if (!processedData.canProcess()) {
+        continue;
+      }
+
+      if (previous == null) {
+        previous = processedData;
+      }
+
+      int maxBatchSize = processedData.getConfig().getInt(MAX_BATCH_SIZE_CONFIG);
+      if (maxBatchSize > 0 && currentGroup.size() == maxBatchSize
+          || !previous.getSinkRecord().topic().equals(processedData.getSinkRecord().topic())
+          || !previous.getNamespace().equals(processedData.getNamespace())) {
+
+        orderedProcessedSinkRecordData.add(currentGroup);
+        currentGroup = new ArrayList<>();
+      }
+      previous = processedData;
+      currentGroup.add(processedData);
+    }
+
+    if (!currentGroup.isEmpty()) {
+      orderedProcessedSinkRecordData.add(currentGroup);
+    }
+    return orderedProcessedSinkRecordData;
+  }
+
+  private MongoSinkRecordProcessor() {}
+}