Exclude updateDescription field from Mongo/DocDB change stream (opens…

…earch-project#4516) Signed-off-by: Dinu John <[email protected]>
dinujoh · May 8, 2024 · f23972c · f23972c
1 parent ad1ab2c
commit f23972c
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 6 deletions.
diff --git a/...s/mongodb/src/main/java/org/opensearch/dataprepper/plugins/mongo/stream/StreamWorker.java b/...s/mongodb/src/main/java/org/opensearch/dataprepper/plugins/mongo/stream/StreamWorker.java
@@ -5,6 +5,8 @@
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoCursor;
 import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.Aggregates;
+import com.mongodb.client.model.Projections;
 import com.mongodb.client.model.changestream.ChangeStreamDocument;
 import com.mongodb.client.model.changestream.FullDocument;
 import com.mongodb.client.model.changestream.OperationType;
@@ -52,6 +54,7 @@ public class StreamWorker {
     static final String FAILURE_ITEM_COUNTER_NAME = "changeEventsProcessingErrors";
     static final String BYTES_RECEIVED = "bytesReceived";
     private static final long MILLI_SECOND = 1_000_000L;
+    private static final String UPDATE_DESCRIPTION = "updateDescription";
     private final RecordBufferWriter recordBufferWriter;
     private final PartitionKeyRecordConverter recordConverter;
     private final DataStreamPartitionCheckpoint partitionCheckpoint;
@@ -124,7 +127,9 @@ public StreamWorker(final RecordBufferWriter recordBufferWriter,
     private MongoCursor<ChangeStreamDocument<Document>> getChangeStreamCursor(final MongoCollection<Document> collection,
                             final String resumeToken
                             ) {
-        final ChangeStreamIterable<Document> changeStreamIterable = collection.watch().batchSize(streamBatchSize);
+        final ChangeStreamIterable<Document> changeStreamIterable = collection.watch(
+                        List.of(Aggregates.project(Projections.exclude(UPDATE_DESCRIPTION))))
+                .batchSize(streamBatchSize);
 
         if (resumeToken == null) {
             return changeStreamIterable.fullDocument(FullDocument.UPDATE_LOOKUP).iterator();

diff --git a/...ngodb/src/test/java/org/opensearch/dataprepper/plugins/mongo/stream/StreamWorkerTest.java b/...ngodb/src/test/java/org/opensearch/dataprepper/plugins/mongo/stream/StreamWorkerTest.java
@@ -60,6 +60,7 @@
 import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
@@ -135,7 +136,7 @@ void test_processStream_success() {
         MongoCursor cursor = mock(MongoCursor.class);
         when(mongoClient.getDatabase(anyString())).thenReturn(mongoDatabase);
         when(mongoDatabase.getCollection(anyString())).thenReturn(col);
-        when(col.watch()).thenReturn(changeStreamIterable);
+        when(col.watch(anyList())).thenReturn(changeStreamIterable);
         when(changeStreamIterable.batchSize(1000)).thenReturn(changeStreamIterable);
         when(changeStreamIterable.fullDocument(FullDocument.UPDATE_LOOKUP)).thenReturn(changeStreamIterable);
         when(changeStreamIterable.iterator()).thenReturn(cursor);
@@ -232,7 +233,7 @@ void test_processStream_checkPointIntervalSuccess() {
         MongoCursor cursor = mock(MongoCursor.class);
         when(mongoClient.getDatabase(anyString())).thenReturn(mongoDatabase);
         when(mongoDatabase.getCollection(anyString())).thenReturn(col);
-        when(col.watch()).thenReturn(changeStreamIterable);
+        when(col.watch(anyList())).thenReturn(changeStreamIterable);
         when(changeStreamIterable.batchSize(1000)).thenReturn(changeStreamIterable);
         when(changeStreamIterable.fullDocument(FullDocument.UPDATE_LOOKUP)).thenReturn(changeStreamIterable);
         when(changeStreamIterable.iterator()).thenReturn(cursor);
@@ -323,7 +324,7 @@ void test_processStream_stopWorker() {
         MongoCursor cursor = mock(MongoCursor.class);
         when(mongoClient.getDatabase(anyString())).thenReturn(mongoDatabase);
         when(mongoDatabase.getCollection(anyString())).thenReturn(col);
-        when(col.watch()).thenReturn(changeStreamIterable);
+        when(col.watch(anyList())).thenReturn(changeStreamIterable);
         when(changeStreamIterable.batchSize(1000)).thenReturn(changeStreamIterable);
         when(changeStreamIterable.fullDocument(FullDocument.UPDATE_LOOKUP)).thenReturn(changeStreamIterable);
         when(changeStreamIterable.iterator()).thenReturn(cursor);
@@ -361,7 +362,7 @@ void test_processStream_terminateChangeStreamSuccess() {
         MongoCursor cursor = mock(MongoCursor.class);
         when(mongoClient.getDatabase(anyString())).thenReturn(mongoDatabase);
         when(mongoDatabase.getCollection(anyString())).thenReturn(col);
-        when(col.watch()).thenReturn(changeStreamIterable);
+        when(col.watch(anyList())).thenReturn(changeStreamIterable);
         when(changeStreamIterable.batchSize(1000)).thenReturn(changeStreamIterable);
         when(changeStreamIterable.fullDocument(FullDocument.UPDATE_LOOKUP)).thenReturn(changeStreamIterable);
         when(changeStreamIterable.iterator()).thenReturn(cursor);
@@ -430,7 +431,7 @@ void test_processStream_dataTypeConversionSuccess(final String actualDocument, f
         MongoCursor cursor = mock(MongoCursor.class);
         when(mongoClient.getDatabase(anyString())).thenReturn(mongoDatabase);
         when(mongoDatabase.getCollection(anyString())).thenReturn(col);
-        when(col.watch()).thenReturn(changeStreamIterable);
+        when(col.watch(anyList())).thenReturn(changeStreamIterable);
         when(changeStreamIterable.batchSize(1000)).thenReturn(changeStreamIterable);
         when(changeStreamIterable.fullDocument(FullDocument.UPDATE_LOOKUP)).thenReturn(changeStreamIterable);
         when(changeStreamIterable.iterator()).thenReturn(cursor);