use batchSize param in MongoReadJournal also for aggregation pipeline…

… batchSize in order to reduce unneeded DB roundtrips

Signed-off-by: Thomas Jaeckle <thomas.jaeckle@bosch.io>

internal/utils/persistence/src/main/java/org/eclipse/ditto/internal/utils/persistence/mongo/streaming/MongoReadJournal.java

@@ -673,7 +673,9 @@ private static Source<Document, NotUsed> listLatestJournalEntries(final MongoCol
                         MongoReadJournal.MAX_BACK_OFF_DURATION, randomFactor)
                 .withMaxRestarts(maxRestarts, minBackOff);
         return RestartSource.onFailuresWithBackoff(restartSettings, () ->
-                Source.fromPublisher(journal.aggregate(pipeline))
+                Source.fromPublisher(journal.aggregate(pipeline)
+                        .batchSize(batchSize) // use batchSize also for the cursor batchSize (16 by default bc of backpressure!)
+                )
         );
     }
 
@@ -750,7 +752,9 @@ private static Source<SnapshotBatch, NotUsed> listNewestActiveSnapshotsByBatch(
             ));
         }
 
-        return Source.fromPublisher(snapshotStore.aggregate(pipeline))
+        return Source.fromPublisher(snapshotStore.aggregate(pipeline)
+                        .batchSize(batchSize) // use batchSize also for the cursor batchSize (16 by default bc of backpressure!)
+                )
                 .flatMapConcat(document -> {
                     final String theMaxPid = document.getString(maxPid);
                     if (theMaxPid == null) {