CDC: Remove oplog $v field from update

KAFKA-44

Author: rozza

src/main/java/com/mongodb/kafka/connect/sink/cdc/debezium/mongodb/MongoDbUpdate.java

@@ -37,6 +37,7 @@
 public class MongoDbUpdate implements CdcOperation {
     private static final ReplaceOptions REPLACE_OPTIONS = new ReplaceOptions().upsert(true);
     private static final String JSON_DOC_FIELD_PATH = "patch";
+    public static final String INTERNAL_OPLOG_FIELD_V = "$v";
 
     @Override
     public WriteModel<BsonDocument> perform(final SinkDocument doc) {
@@ -47,6 +48,13 @@ public WriteModel<BsonDocument> perform(final SinkDocument doc) {
 
         try {
             BsonDocument updateDoc = BsonDocument.parse(valueDoc.getString(JSON_DOC_FIELD_PATH).getValue());
+
+            //Check if the internal "$v" field is contained which was added to the
+            //oplog format in 3.6+ If so, then we simply remove it for now since
+            //it's not used by the sink connector at the moment and would break
+            //CDC-mode based "replication".
+            updateDoc.remove(INTERNAL_OPLOG_FIELD_V);
+
             //patch contains full new document for replacement
             if (updateDoc.containsKey(ID_FIELD)) {
                 BsonDocument filterDoc = new BsonDocument(ID_FIELD, updateDoc.get(ID_FIELD));

src/test/java/com/mongodb/kafka/connect/sink/cdc/debezium/mongodb/MongoDbUpdateTest.java

@@ -29,6 +29,7 @@
 import org.junit.runner.RunWith;
 
 import org.bson.BsonDocument;
+import org.bson.BsonInt32;
 import org.bson.BsonString;
 
 import com.mongodb.client.model.ReplaceOneModel;
@@ -43,6 +44,7 @@ class MongoDbUpdateTest {
     private static final BsonDocument FILTER_DOC = BsonDocument.parse("{_id: 1234}");
     private static final BsonDocument REPLACEMENT_DOC = BsonDocument.parse("{_id: 1234, first_name: 'Grace', last_name: 'Hopper'}");
     private static final BsonDocument UPDATE_DOC = BsonDocument.parse("{$set: {first_name: 'Grace', last_name: 'Hopper'}}");
+    private static final BsonDocument UPDATE_DOC_WITH_OPLOG_INTERNALS = UPDATE_DOC.clone().append("$v", new BsonInt32(1));
 
     @Test
     @DisplayName("when valid doc replace cdc event then correct ReplaceOneModel")
@@ -79,6 +81,22 @@ void testValidSinkDocumentForUpdate() {
         assertEquals(FILTER_DOC, writeModel.getFilter());
     }
 
+    @Test
+    @DisplayName("when valid doc change cdc event containing internal oplog fields then correct UpdateOneModel")
+    public void testValidSinkDocumentWithInternalOploagFieldForUpdate() {
+        BsonDocument keyDoc = BsonDocument.parse("{id: '1234'}");
+        BsonDocument valueDoc = new BsonDocument("op", new BsonString("u"))
+                .append("patch", new BsonString(UPDATE_DOC_WITH_OPLOG_INTERNALS.toJson()));
+
+        WriteModel<BsonDocument> result = UPDATE.perform(new SinkDocument(keyDoc, valueDoc));
+        assertTrue(result instanceof UpdateOneModel, () -> "result expected to be of type UpdateOneModel");
+
+        UpdateOneModel<BsonDocument> writeModel = (UpdateOneModel<BsonDocument>) result;
+        assertEquals(UPDATE_DOC, writeModel.getUpdate(), () -> "update doc not matching what is expected");
+        assertTrue(writeModel.getFilter() instanceof BsonDocument, () -> "filter expected to be of type BsonDocument");
+        assertEquals(FILTER_DOC, writeModel.getFilter());
+    }
+
     @Test
     @DisplayName("when missing value doc then DataException")
     void testMissingValueDocument() {