Added UpdateOneDefaultStrategy (#144)

A WriteStrategy that performs a `$set` upsert operation.
For scenarios where the topic data contains fragments of the
desired data in MongoDB

KAFKA-255

Author: rozza

src/main/java/com/mongodb/kafka/connect/sink/writemodel/strategy/UpdateOneDefaultStrategy.java

@@ -0,0 +1,57 @@
+/*
+ * 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.
+ *
+ * Original Work: Apache License, Version 2.0, Copyright 2017 Hans-Peter Grahsl.
+ */
+
+package com.mongodb.kafka.connect.sink.writemodel.strategy;
+
+import static com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.ID_FIELD;
+
+import org.apache.kafka.connect.errors.DataException;
+
+import org.bson.BsonDocument;
+import org.bson.BsonValue;
+
+import com.mongodb.client.model.UpdateOneModel;
+import com.mongodb.client.model.UpdateOptions;
+import com.mongodb.client.model.WriteModel;
+
+import com.mongodb.kafka.connect.sink.converter.SinkDocument;
+
+public class UpdateOneDefaultStrategy implements WriteModelStrategy {
+
+  private static final UpdateOptions UPDATE_OPTIONS = new UpdateOptions().upsert(true);
+
+  @Override
+  public WriteModel<BsonDocument> createWriteModel(final SinkDocument document) {
+    BsonDocument vd =
+        document
+            .getValueDoc()
+            .orElseThrow(
+                () ->
+                    new DataException(
+                        "Could not build the WriteModel,the value document was missing unexpectedly"));
+
+    BsonValue idValue = vd.get(ID_FIELD);
+    if (idValue == null) {
+      throw new DataException(
+          "Could not build the WriteModel,the `_id` field was missing unexpectedly");
+    }
+    vd.remove(ID_FIELD);
+    return new UpdateOneModel<>(
+        new BsonDocument(ID_FIELD, idValue), new BsonDocument("$set", vd), UPDATE_OPTIONS);
+  }
+}

src/test/java/com/mongodb/kafka/connect/sink/writemodel/strategy/WriteModelStrategyTest.java

@@ -56,6 +56,8 @@ class WriteModelStrategyTest {
       new ReplaceOneDefaultStrategy();
   private static final ReplaceOneBusinessKeyStrategy REPLACE_ONE_BUSINESS_KEY_STRATEGY =
       new ReplaceOneBusinessKeyStrategy();
+  private static final UpdateOneDefaultStrategy UPDATE_ONE_DEFAULT_STRATEGY =
+      new UpdateOneDefaultStrategy();
   private static final UpdateOneTimestampsStrategy UPDATE_ONE_TIMESTAMPS_STRATEGY =
       new UpdateOneTimestampsStrategy();
   private static final UpdateOneBusinessKeyTimestampStrategy
@@ -231,6 +233,31 @@ void testReplaceOneBusinessKeyStrategyPartialWithValidSinkDocument() {
         "replacement expected to be done in upsert mode");
   }
 
+  @Test
+  @DisplayName(
+      "when sink document is valid for UpdateOneDefaultStrategy then correct UpdateOneModel")
+  void testUpdateOneDefaultStrategyWithValidSinkDocument() {
+    WriteModel<BsonDocument> result =
+        UPDATE_ONE_DEFAULT_STRATEGY.createWriteModel(new SinkDocument(null, VALUE_DOC.clone()));
+    assertTrue(result instanceof UpdateOneModel, "result expected to be of type UpdateOneModel");
+
+    UpdateOneModel<BsonDocument> writeModel = (UpdateOneModel<BsonDocument>) result;
+
+    BsonDocument setValuesDocument = VALUE_DOC.clone();
+    setValuesDocument.remove("_id");
+    BsonDocument expectedSetDocument = new BsonDocument("$set", setValuesDocument);
+
+    assertEquals(
+        expectedSetDocument,
+        writeModel.getUpdate(),
+        "replacement doc not matching what is expected");
+    assertTrue(
+        writeModel.getFilter() instanceof BsonDocument,
+        "filter expected to be of type BsonDocument");
+    assertEquals(ID_FILTER, writeModel.getFilter());
+    assertTrue(writeModel.getOptions().isUpsert(), "update expected to be done in upsert mode");
+  }
+
   @Test
   @DisplayName(
       "when sink document is valid for UpdateOneTimestampsStrategy then correct UpdateOneModel")
@@ -412,6 +439,10 @@ void testIEmptyOrMissingSinkDocumentData() {
                 () ->
                     REPLACE_ONE_BUSINESS_KEY_PARTIAL_STRATEGY.createWriteModel(
                         SINK_DOCUMENT_EMPTY)),
+        () ->
+            assertThrows(
+                DataException.class,
+                () -> UPDATE_ONE_DEFAULT_STRATEGY.createWriteModel(SINK_DOCUMENT_EMPTY)),
         () ->
             assertThrows(
                 DataException.class,