Fix code not terminating on repository saving an empty flux.

Original Pull Request #3099
Closes: #3039

Signed-off-by: Peter-Josef Meisch <[email protected]>

Author: sothawo

src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java

@@ -233,6 +233,7 @@ public <T> Flux<T> save(Flux<T> entities, IndexCoordinates index, int bulkSize)
 					.subscribe(new Subscriber<>() {
 						@Nullable private Subscription subscription = null;
 						private final AtomicBoolean upstreamComplete = new AtomicBoolean(false);
+						private final AtomicBoolean onNextHasBeenCalled = new AtomicBoolean(false);
 
 						@Override
 						public void onSubscribe(Subscription subscription) {
@@ -242,6 +243,7 @@ public void onSubscribe(Subscription subscription) {
 
 						@Override
 						public void onNext(List<T> entityList) {
+							onNextHasBeenCalled.set(true);
 							saveAll(entityList, index)
 									.map(sink::tryEmitNext)
 									.doOnComplete(() -> {
@@ -267,6 +269,10 @@ public void onError(Throwable throwable) {
 						@Override
 						public void onComplete() {
 							upstreamComplete.set(true);
+							if (!onNextHasBeenCalled.get()) {
+								// this happens when an empty flux is saved
+								sink.tryEmitComplete();
+							}
 						}
 					});
 			return sink.asFlux();

src/test/java/org/springframework/data/elasticsearch/repository/support/SimpleReactiveElasticsearchRepositoryIntegrationTests.java

@@ -105,6 +105,26 @@ private Mono<Boolean> documentWithIdExistsInIndex(String id) {
 		return operations.exists(id, IndexCoordinates.of(indexNameProvider.indexName()));
 	}
 
+	@Test // #3093
+	@DisplayName("should save all from empty collection")
+	void shouldSaveAllFromEmptyCollection() {
+
+		repository.saveAll(Collections.emptyList())
+				.as(StepVerifier::create)
+				.expectNextCount(0)
+				.verifyComplete();
+	}
+
+	@Test // #3093
+	@DisplayName("should save all from empty flux")
+	void shouldSaveAllFromEmptyFlux() {
+
+		repository.saveAll(Flux.empty())
+				.as(StepVerifier::create)
+				.expectNextCount(0)
+				.verifyComplete();
+	}
+
 	@Test // DATAES-519
 	void saveShouldComputeMultipleEntities() {