Switch to Flux.flatMapSequential(…) to prevent backpressure shaping

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>4.2.0-SNAPSHOT</version>
+	<version>4.2.0-GH-4543-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.2.0-SNAPSHOT</version>
+		<version>4.2.0-GH-4543-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-distribution/pom.xml

@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.2.0-SNAPSHOT</version>
+		<version>4.2.0-GH-4543-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -13,7 +13,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.2.0-SNAPSHOT</version>
+		<version>4.2.0-GH-4543-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultReactiveBulkOperations.java

@@ -216,7 +216,7 @@ private Mono<BulkWriteResult> bulkWriteTo(MongoCollection<Document> collection)
 			collection = collection.withWriteConcern(defaultWriteConcern);
 		}
 
-		Flux<SourceAwareWriteModelHolder> concat = Flux.concat(models).flatMap(it -> {
+		Flux<SourceAwareWriteModelHolder> concat = Flux.concat(models).flatMapSequential(it -> {
 
 			if (it.model()instanceof InsertOneModel<Document> iom) {
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -1051,7 +1051,7 @@ private <O> Flux<O> aggregateAndMap(MongoCollection<Document> collection, List<D
 			return (isOutOrMerge ? Flux.from(cursor.toCollection()) : Flux.from(cursor.first())).thenMany(Mono.empty());
 		}
 
-		return Flux.from(cursor).concatMap(readCallback::doWith);
+		return Flux.from(cursor).flatMapSequential(readCallback::doWith);
 	}
 
 	@Override
@@ -1098,7 +1098,7 @@ protected <T> Flux<GeoResult<T>> geoNear(NearQuery near, Class<?> entityClass, S
 				.withOptions(optionsBuilder.build());
 
 		return aggregate($geoNear, collection, Document.class) //
-				.concatMap(callback::doWith);
+				.flatMapSequential(callback::doWith);
 	}
 
 	@Override
@@ -1324,7 +1324,7 @@ public <T> Flux<T> insertAll(Mono<? extends Collection<? extends T>> batchToSave
 
 		Assert.notNull(batchToSave, "Batch to insert must not be null");
 
-		return Flux.from(batchToSave).flatMap(collection -> insert(collection, collectionName));
+		return Flux.from(batchToSave).flatMapSequential(collection -> insert(collection, collectionName));
 	}
 
 	@Override
@@ -1392,7 +1392,7 @@ public <T> Flux<T> insertAll(Collection<? extends T> objectsToSave) {
 
 	@Override
 	public <T> Flux<T> insertAll(Mono<? extends Collection<? extends T>> objectsToSave) {
-		return Flux.from(objectsToSave).flatMap(this::insertAll);
+		return Flux.from(objectsToSave).flatMapSequential(this::insertAll);
 	}
 
 	protected <T> Flux<T> doInsertAll(Collection<? extends T> listToSave, MongoWriter<Object> writer) {
@@ -1443,7 +1443,7 @@ protected <T> Flux<T> doInsertBatch(String collectionName, Collection<? extends
 			return insertDocumentList(collectionName, documents).thenMany(Flux.fromIterable(tuples));
 		});
 
-		return insertDocuments.flatMap(tuple -> {
+		return insertDocuments.flatMapSequential(tuple -> {
 
 			Document document = tuple.getT2();
 			Object id = MappedDocument.of(document).getId();
@@ -1600,7 +1600,7 @@ protected Flux<ObjectId> insertDocumentList(String collectionName, List<Document
 
 			return collectionToUse.insertMany(documents);
 
-		}).flatMap(s -> {
+		}).flatMapSequential(s -> {
 
 			return Flux.fromStream(documents.stream() //
 					.map(MappedDocument::of) //
@@ -2187,7 +2187,7 @@ public <T> Flux<T> mapReduce(Query filterQuery, Class<?> domainType, String inpu
 			publisher = collation.map(Collation::toMongoCollation).map(publisher::collation).orElse(publisher);
 
 			return Flux.from(publisher)
-					.concatMap(new ReadDocumentCallback<>(mongoConverter, resultType, inputCollectionName)::doWith);
+					.flatMapSequential(new ReadDocumentCallback<>(mongoConverter, resultType, inputCollectionName)::doWith);
 		});
 	}
 
@@ -2255,7 +2255,7 @@ protected <T> Flux<T> doFindAndDelete(String collectionName, Query query, Class<
 
 		return Flux.from(flux).collectList().filter(it -> !it.isEmpty())
 				.flatMapMany(list -> Flux.from(remove(operations.getByIdInQuery(list), entityClass, collectionName))
-						.flatMap(deleteResult -> Flux.fromIterable(list)));
+						.flatMapSequential(deleteResult -> Flux.fromIterable(list)));
 	}
 
 	/**
@@ -2729,7 +2729,7 @@ private <T> Flux<T> executeFindMultiInternal(ReactiveCollectionQueryCallback<Doc
 
 		return createFlux(collectionName, collection -> {
 			return Flux.from(preparer.initiateFind(collection, collectionCallback::doInCollection))
-					.concatMap(objectCallback::doWith);
+					.flatMapSequential(objectCallback::doWith);
 		});
 	}
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/SimpleReactiveMongoRepository.java

@@ -121,7 +121,7 @@ public <S extends T> Flux<S> saveAll(Publisher<S> entityStream) {
 
 		Assert.notNull(entityStream, "The given Publisher of entities must not be null");
 
-		return Flux.from(entityStream).flatMap(entity -> entityInformation.isNew(entity) ? //
+		return Flux.from(entityStream).flatMapSequential(entity -> entityInformation.isNew(entity) ? //
 				mongoOperations.insert(entity, entityInformation.getCollectionName()) : //
 				mongoOperations.save(entity, entityInformation.getCollectionName()));
 	}
@@ -191,7 +191,7 @@ public Flux<T> findAllById(Publisher<ID> ids) {
 		Assert.notNull(ids, "The given Publisher of Id's must not be null");
 
 		Optional<ReadPreference> readPreference = getReadPreference();
-		return Flux.from(ids).buffer().flatMap(listOfIds -> {
+		return Flux.from(ids).buffer().flatMapSequential(listOfIds -> {
 			Query query = getIdQuery(listOfIds);
 			readPreference.ifPresent(query::withReadPreference);
 			return mongoOperations.find(query, entityInformation.getJavaType(), entityInformation.getCollectionName());
@@ -345,7 +345,8 @@ public <S extends T> Flux<S> insert(Publisher<S> entities) {
 
 		Assert.notNull(entities, "The given Publisher of entities must not be null");
 
-		return Flux.from(entities).flatMap(entity -> mongoOperations.insert(entity, entityInformation.getCollectionName()));
+		return Flux.from(entities)
+				.flatMapSequential(entity -> mongoOperations.insert(entity, entityInformation.getCollectionName()));
 	}
 
 	// -------------------------------------------------------------------------

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateUnitTests.java

@@ -34,6 +34,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
 import org.assertj.core.api.Assertions;
@@ -697,6 +698,28 @@ void aggreateShouldUseReadReadPreference() {
 		verify(collection).withReadPreference(ReadPreference.primaryPreferred());
 	}
 
+	@Test // GH-4543
+	void aggregateDoesNotLimitBackpressure() {
+
+		reset(collection);
+
+		AtomicLong request = new AtomicLong();
+		Publisher<Document> realPublisher = Flux.just(new Document()).doOnRequest(request::addAndGet);
+
+		doAnswer(invocation -> {
+			Subscriber<Document> subscriber = invocation.getArgument(0);
+			realPublisher.subscribe(subscriber);
+			return null;
+		}).when(aggregatePublisher).subscribe(any());
+
+		when(collection.aggregate(anyList())).thenReturn(aggregatePublisher);
+		when(collection.aggregate(anyList(), any(Class.class))).thenReturn(aggregatePublisher);
+
+		template.aggregate(newAggregation(Sith.class, project("id")), AutogenerateableId.class, Document.class).subscribe();
+
+		assertThat(request).hasValueGreaterThan(128);
+	}
+
 	@Test // DATAMONGO-1854
 	void aggreateShouldUseCollationFromOptionsEvenIfDefaultCollationIsPresent() {
 
@@ -1261,6 +1284,17 @@ void findShouldInvokeAfterConvertCallbacks() {
 		assertThat(results.get(0).id).isEqualTo("after-convert");
 	}
 
+	@Test // GH-4543
+	void findShouldNotLimitBackpressure() {
+
+		AtomicLong request = new AtomicLong();
+		stubFindSubscribe(new Document(), request);
+
+		template.find(new Query(), Person.class).subscribe();
+
+		assertThat(request).hasValueGreaterThan(128);
+	}
+
 	@Test // DATAMONGO-2479
 	void findByIdShouldInvokeAfterConvertCallbacks() {
 
@@ -1706,8 +1740,12 @@ public WriteConcern resolve(MongoAction action) {
 	}
 
 	private void stubFindSubscribe(Document document) {
+		stubFindSubscribe(document, new AtomicLong());
+	}
+
+	private void stubFindSubscribe(Document document, AtomicLong request) {
 
-		Publisher<Document> realPublisher = Flux.just(document);
+		Publisher<Document> realPublisher = Flux.just(document).doOnRequest(request::addAndGet);
 
 		doAnswer(invocation -> {
 			Subscriber<Document> subscriber = invocation.getArgument(0);