Fixed using ns field in copy.existing pipeline bug

KAFKA-217

Author: rozza

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 ### Bug Fixes
   - [KAFKA-227](https://jira.mongodb.org/browse/KAFKA-195) Fixed wrapping nullable value returned from WriteModelStrategy
+  - [KAFKA-217](https://jira.mongodb.org/browse/KAFKA-217) Fixed using `ns` field in `copy.existing` pipeline bug
 
 ## 1.5.0
 

src/main/java/com/mongodb/kafka/connect/source/MongoCopyDataManager.java

@@ -65,8 +65,9 @@
  */
 class MongoCopyDataManager implements AutoCloseable {
   private static final Logger LOGGER = LoggerFactory.getLogger(MongoCopyDataManager.class);
-  static final String NAMESPACE_FIELD = "__";
-  private static final byte[] NAMESPACE_BYTES = "ns".getBytes(StandardCharsets.UTF_8);
+  private static final String NAMESPACE_FIELD = "ns";
+  static final String ALT_NAMESPACE_FIELD = "__";
+  private static final byte[] NAMESPACE_BYTES = NAMESPACE_FIELD.getBytes(StandardCharsets.UTF_8);
 
   private static final String PIPELINE_TEMPLATE =
       format(
@@ -80,6 +81,12 @@ class MongoCopyDataManager implements AutoCloseable {
               + "}",
           NAMESPACE_FIELD);
 
+  private static final BsonDocument ADD_ALT_NAMESPACE_STAGE =
+      BsonDocument.parse(
+          format("{'$addFields': {'%s': '$%s'}}", ALT_NAMESPACE_FIELD, NAMESPACE_FIELD));
+  private static final BsonDocument UNSET_ORIGINAL_NAMESPACE_STAGE =
+      BsonDocument.parse(format("{'$project': {'%s': 0}}", NAMESPACE_FIELD));
+
   private volatile boolean closed;
   private volatile Exception errorException;
   private final AtomicInteger namespacesToCopy;
@@ -187,6 +194,8 @@ static List<Bson> createPipeline(final MongoSourceConfig cfg, final MongoNamespa
         BsonDocument.parse(
             format(PIPELINE_TEMPLATE, namespace.getDatabaseName(), namespace.getCollectionName())));
     cfg.getPipeline().map(pipeline::addAll);
+    pipeline.add(ADD_ALT_NAMESPACE_STAGE);
+    pipeline.add(UNSET_ORIGINAL_NAMESPACE_STAGE);
     return pipeline;
   }
 
@@ -196,7 +205,7 @@ static RawBsonDocument convertDocument(final RawBsonDocument original) {
     int currentPosition = 0;
     reader.readStartDocument();
     while (reader.readBsonType() != BsonType.END_OF_DOCUMENT) {
-      if (reader.readName().equals(NAMESPACE_FIELD)) {
+      if (reader.readName().equals(ALT_NAMESPACE_FIELD)) {
         currentPosition++; // Adjust the current position to include the bson type
         byte[] sourceBytes = sourceBuffer.array();
         // Convert the namespace field in situ

src/test/java/com/mongodb/kafka/connect/source/MongoCopyDataManagerTest.java

@@ -15,7 +15,7 @@
  */
 package com.mongodb.kafka.connect.source;
 
-import static com.mongodb.kafka.connect.source.MongoCopyDataManager.NAMESPACE_FIELD;
+import static com.mongodb.kafka.connect.source.MongoCopyDataManager.ALT_NAMESPACE_FIELD;
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.COLLECTION_CONFIG;
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.COPY_EXISTING_NAMESPACE_REGEX_CONFIG;
 import static com.mongodb.kafka.connect.source.MongoSourceConfig.COPY_EXISTING_PIPELINE_CONFIG;
@@ -436,7 +436,7 @@ private static String createTemplate(final int id, final String dbName, final St
   }
 
   private static RawBsonDocument createInput(final String json) {
-    return RawBsonDocument.parse(format(json, NAMESPACE_FIELD));
+    return RawBsonDocument.parse(format(json, ALT_NAMESPACE_FIELD));
   }
 
   private static Optional<BsonDocument> createOutput(final String json) {