Add settings to mongodb to configure write batches (#224)

koralium · Dec 24, 2023 · 6a75bfe · 6a75bfe
1 parent 21899b9
commit 6a75bfe
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 3 deletions.
diff --git a/src/FlowtideDotNet.Connector.MongoDB/FlowtideMongoDBSinkOptions.cs b/src/FlowtideDotNet.Connector.MongoDB/FlowtideMongoDBSinkOptions.cs
@@ -36,5 +36,15 @@ public class FlowtideMongoDBSinkOptions
         public Func<IMongoCollection<BsonDocument>, Task>? OnInitialDataSent { get; set; }
 
         public Func<Watermark, Task>? OnWatermarkUpdate { get; set; }
+
+        /// <summary>
+        /// Set the amount of documents that will be sent per batch to mongodb.
+        /// </summary>
+        public int DocumentsPerBatch { get; set; } = 100;
+
+        /// <summary>
+        /// Set the amount of batches that will be sent in parallel to mongodb.
+        /// </summary>
+        public int ParallelBatches { get; set; } = 10;
     }
 }
diff --git a/src/FlowtideDotNet.Connector.MongoDB/Internal/MongoDBSink.cs b/src/FlowtideDotNet.Connector.MongoDB/Internal/MongoDBSink.cs
@@ -155,9 +155,9 @@ protected override async Task UploadChanges(IAsyncEnumerable<SimpleChangeEvent>
                     writes.Add(new ReplaceOneModel<BsonDocument>(filter, doc) { IsUpsert = true });
                 }
 
-                if (writes.Count >= 100)
+                if (writes.Count >= options.DocumentsPerBatch)
                 {
-                    while (writeTasks.Count >= 100)
+                    while (writeTasks.Count >= options.ParallelBatches)
                     {
                         for(int i = 0; i < writeTasks.Count; i++)
                         {
@@ -178,7 +178,7 @@ protected override async Task UploadChanges(IAsyncEnumerable<SimpleChangeEvent>
                                 writeTasks.RemoveAt(i);
                             }
                         }
-                        if (writeTasks.Count >= 100)
+                        if (writeTasks.Count >= options.ParallelBatches)
                         {
                             await Task.WhenAny(writeTasks);
                         }