merge substreams before executing bulkWrite

Signed-off-by: Thomas Jaeckle <thomas.jaeckle@bosch.io>
eclipse-ditto · Mar 10, 2022 · 1c66d98 · 1c66d98
1 parent c707120
commit 1c66d98
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...eclipse/ditto/thingsearch/service/persistence/write/streaming/MongoSearchUpdaterFlow.java b/...eclipse/ditto/thingsearch/service/persistence/write/streaming/MongoSearchUpdaterFlow.java
@@ -112,8 +112,8 @@ public Flow<SubSource<AbstractWriteModel, NotUsed>, WriteResultAndErrors, NotUse
         return Flow.<SubSource<AbstractWriteModel, NotUsed>>create()
                 .flatMapConcat(source -> source.grouped(maxBulkSize)
                         .flatMapConcat(searchUpdateMapper::processWriteModels)
-                        .flatMapMerge(parallelism, writeModels -> executeBulkWrite(shouldAcknowledge, writeModels))
                         .mergeSubstreamsWithParallelism(maxBulkSize)
+                        .flatMapMerge(parallelism, writeModels -> executeBulkWrite(shouldAcknowledge, writeModels))
                         .async(MongoSearchUpdaterFlow.DISPATCHER_NAME, parallelism)
                 );
     }