Fixed bug in LazyBsonDocument#clone ignoring any changes made once unwrapped.

KAFKA-218

Author: rozza

CHANGELOG.md

@@ -10,6 +10,9 @@
     the dependencies needed for running the connector with confluent. `mongo-kafka-connect-<version>-all.jar` now also includes `Avro` 
     dependencies for ease of deployment for alternative Kafka connect runtimes.
 
+### Bug Fixes
+  - [KAFKA-218](https://jira.mongodb.org/browse/KAFKA-218) Fixed bug in LazyBsonDocument#clone ignoring any changes made once unwrapped.
+
 ## 1.5.1
 
 ### Bug Fixes

src/main/java/com/mongodb/kafka/connect/sink/converter/LazyBsonDocument.java

@@ -145,7 +145,9 @@ public String toString() {
 
   @Override
   public BsonDocument clone() {
-    return new LazyBsonDocument(record, dataType, converter);
+    return unwrapped != null
+        ? unwrapped.clone()
+        : new LazyBsonDocument(record, dataType, converter);
   }
 
   private BsonDocument getUnwrapped() {

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

@@ -32,6 +32,7 @@
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.errors.DataException;
@@ -46,6 +47,7 @@
 import com.mongodb.MongoNamespace;
 import com.mongodb.client.model.ReplaceOneModel;
 import com.mongodb.client.model.ReplaceOptions;
+import com.mongodb.client.model.UpdateOneModel;
 
 import com.mongodb.kafka.connect.sink.cdc.debezium.mongodb.MongoDbHandler;
 import com.mongodb.kafka.connect.sink.namespace.mapping.FieldPathNamespaceMapper;
@@ -197,6 +199,52 @@ INVALID_SINK_RECORD, createSinkConfig(ERRORS_TOLERANCE_CONFIG, "all"))
                                 FIELD_NAMESPACE_MAPPER_ERROR_IF_INVALID_CONFIG)))));
   }
 
+  @Test
+  @DisplayName("Rename _id handling")
+  void testRenameIdHandling() {
+    SinkRecord sinkRecord =
+        new SinkRecord(
+            TEST_TOPIC,
+            0,
+            Schema.STRING_SCHEMA,
+            "{_id: 1}",
+            Schema.STRING_SCHEMA,
+            "{a: 'a', b: 'b', c: 'c', d: 'd'}",
+            1);
+
+    String topicConfig =
+        "{"
+            + "'writemodel.strategy': 'com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneBusinessKeyTimestampStrategy',"
+            + "'post.processor.chain': 'com.mongodb.kafka.connect.sink.processor.field.renaming.RenameByMapping, "
+            + "com.mongodb.kafka.connect.sink.processor.DocumentIdAdder',"
+            + "'field.renamer.mapping': '[{oldName: \"value.c\", newName: \"_id\"}]',"
+            + "'document.id.strategy': 'com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy',"
+            + "'document.id.strategy.overwrite.existing': 'true',"
+            + "'document.id.strategy.partial.value.projection.type': 'allowlist',"
+            + "'document.id.strategy.partial.value.projection.list': 'a, b, _id'"
+            + "}";
+
+    MongoProcessedSinkRecordData processedData =
+        new MongoProcessedSinkRecordData(sinkRecord, createSinkConfig(topicConfig));
+    assertNull(processedData.getException());
+    UpdateOneModel<BsonDocument> writeModel =
+        (UpdateOneModel<BsonDocument>) processedData.getWriteModel();
+    assertTrue(writeModel.getOptions().isUpsert());
+    assertEquals(BsonDocument.parse("{'a': 'a', 'b': 'b', '_id': 'c'}"), writeModel.getFilter());
+
+    BsonDocument update = (BsonDocument) writeModel.getUpdate();
+    assertNotNull(update);
+    BsonDocument setDocument = update.getDocument("$set", new BsonDocument());
+    assertTrue(setDocument.containsKey("_modifiedTS"));
+    setDocument.remove("_modifiedTS");
+    assertEquals(BsonDocument.parse("{'a': 'a', 'b': 'b', 'd': 'd'}"), setDocument);
+
+    BsonDocument setOnInsertDocument = update.getDocument("$setOnInsert", new BsonDocument());
+    assertTrue(setOnInsertDocument.containsKey("_insertedTS"));
+    setOnInsertDocument.remove("_insertedTS");
+    assertTrue(setOnInsertDocument.isEmpty());
+  }
+
   void assertWriteModel(final MongoProcessedSinkRecordData processedData) {
     assertWriteModel(processedData, EXPECTED_WRITE_MODEL);
   }

src/test/java/com/mongodb/kafka/connect/sink/converter/LazyBsonDocumentTest.java

@@ -49,6 +49,8 @@ public class LazyBsonDocumentTest {
       new SinkRecord("topic", 0, Schema.STRING_SCHEMA, NOT_JSON, Schema.STRING_SCHEMA, JSON, 1L);
   private static final SinkRecord SINK_RECORD_ALT_VALUE =
       new SinkRecord("topic", 0, Schema.STRING_SCHEMA, JSON, Schema.STRING_SCHEMA, NOT_JSON, 1L);
+  private static final SinkRecord VALID_SINK_RECORD =
+      new SinkRecord("topic", 0, Schema.STRING_SCHEMA, JSON, Schema.STRING_SCHEMA, JSON, 1L);
 
   @Test
   @DisplayName("test invalid LazyBsonDocuments")
@@ -154,6 +156,16 @@ void testClone() {
                   Type.VALUE,
                   (Schema schema, Object data) -> BsonDocument.parse(data.toString()));
           assertDoesNotThrow(lazyBsonDocument::clone);
+        },
+        () -> {
+          LazyBsonDocument lazyBsonDocument =
+              new LazyBsonDocument(
+                  VALID_SINK_RECORD,
+                  Type.VALUE,
+                  (Schema schema, Object data) -> BsonDocument.parse(data.toString()));
+          lazyBsonDocument.remove("_id");
+          assertEquals(lazyBsonDocument, lazyBsonDocument.clone());
+          assertEquals(BsonDocument.parse("{a: 'a', b: 'b', c: 'c'}"), lazyBsonDocument.clone());
         });
   }
 }