Adding DeleteExactKeyStrategy

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

@@ -0,0 +1,44 @@
+/*
+ * 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 org.apache.kafka.connect.errors.DataException;
+
+import org.bson.BsonDocument;
+
+import com.mongodb.client.model.DeleteOneModel;
+import com.mongodb.client.model.WriteModel;
+
+import com.mongodb.kafka.connect.sink.converter.SinkDocument;
+
+public class DeleteExactKeyStrategy implements WriteModelStrategy {
+
+  @Override
+  public WriteModel<BsonDocument> createWriteModel(final SinkDocument document) {
+    BsonDocument kd =
+        document
+            .getKeyDoc()
+            .orElseThrow(
+                () ->
+                    new DataException(
+                        "Could not build the WriteModel,the key document was missing unexpectedly"));
+
+    return new DeleteOneModel<>(kd);
+  }
+}

src/test/java/com/mongodb/kafka/connect/sink/MongoSinkConfigTest.java

@@ -106,6 +106,7 @@
 import com.mongodb.kafka.connect.sink.processor.id.strategy.UuidStrategy;
 import com.mongodb.kafka.connect.sink.writemodel.strategy.CustomDeleteWriteModelStrategy;
 import com.mongodb.kafka.connect.sink.writemodel.strategy.DefaultWriteModelStrategy;
+import com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteExactKeyStrategy;
 import com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneBusinessKeyStrategy;
 import com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneDefaultStrategy;
 import com.mongodb.kafka.connect.sink.writemodel.strategy.InsertOneDefaultStrategy;
@@ -712,6 +713,7 @@ Collection<DynamicTest> testGetSingleValidWriteModelStrategy() {
                 UpdateOneBusinessKeyTimestampStrategy.class.getName(),
                 UpdateOneBusinessKeyTimestampStrategy.class);
             put(DeleteOneBusinessKeyStrategy.class.getName(), DeleteOneBusinessKeyStrategy.class);
+            put(DeleteExactKeyStrategy.class.getName(), DeleteExactKeyStrategy.class);
           }
         };
 
@@ -796,6 +798,7 @@ Collection<DynamicTest> testGetSingleValidDeleteWriteModelStrategy() {
                 UpdateOneBusinessKeyTimestampStrategy.class.getName(),
                 UpdateOneBusinessKeyTimestampStrategy.class);
             put(DeleteOneBusinessKeyStrategy.class.getName(), DeleteOneBusinessKeyStrategy.class);
+            put(DeleteExactKeyStrategy.class.getName(), DeleteExactKeyStrategy.class);
           }
         };
 

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

@@ -68,6 +68,8 @@ class WriteModelStrategyTest {
   private static final DeleteOneBusinessKeyStrategy DELETE_ONE_BUSINESS_KEY_STRATEGY =
       new DeleteOneBusinessKeyStrategy();
   private static final DeleteOneBusinessKeyStrategy DELETE_ONE_BUSINESS_KEY_PARTIAL_STRATEGY;
+  private static final DeleteExactKeyStrategy DELETE_EXACT_KEY_STRATEGY =
+      new DeleteExactKeyStrategy();
   private static final SinkDocument SINK_DOCUMENT_NULL_VALUE =
       new SinkDocument(new BsonDocument(), null);
   private static final SinkDocument SINK_DOCUMENT_NULL_KEY =
@@ -403,6 +405,25 @@ void testDeleteOneBusinessKeyStrategyStrategyPartialWithValidSinkDocument() {
     assertEquals(BUSINESS_KEY_FLATTENED_FILTER, writeModel.getFilter());
   }
 
+  @Test
+  @DisplayName("when sink document is valid for DeleteExactKeyStrategy then correct DeleteOneModel")
+  void testDeleteExactKeyStrategyWitValidSinkDocument() {
+    BsonDocument keyDoc = BsonDocument.parse("{id: 1234, id2: 4321}");
+
+    WriteModel<BsonDocument> result =
+        DELETE_EXACT_KEY_STRATEGY.createWriteModel(new SinkDocument(keyDoc, null));
+
+    assertTrue(result instanceof DeleteOneModel, "result expected to be of type DeleteOneModel");
+
+    DeleteOneModel<BsonDocument> writeModel = (DeleteOneModel<BsonDocument>) result;
+
+    assertTrue(
+        writeModel.getFilter() instanceof BsonDocument,
+        "filter expected to be of type BsonDocument");
+
+    assertEquals(BsonDocument.parse("{id: 1234, id2: 4321}"), writeModel.getFilter());
+  }
+
   @Test
   @DisplayName("Test handling empty or missing sink document data")
   void testIEmptyOrMissingSinkDocumentData() {
@@ -493,7 +514,10 @@ void testIEmptyOrMissingSinkDocumentData() {
             assertThrows(
                 DataException.class,
                 () ->
-                    DELETE_ONE_BUSINESS_KEY_PARTIAL_STRATEGY.createWriteModel(
-                        SINK_DOCUMENT_EMPTY)));
+                    DELETE_ONE_BUSINESS_KEY_PARTIAL_STRATEGY.createWriteModel(SINK_DOCUMENT_EMPTY)),
+        () ->
+            assertThrows(
+                DataException.class,
+                () -> DELETE_EXACT_KEY_STRATEGY.createWriteModel(SINK_DOCUMENT_NULL_KEY)));
   }
 }