Fix error propagation in AbstractReactiveElasticsearchTemplate:save()

Previously, errors occurring during the saveAll operation within the reactive save method were swallowed because the inner subscriber did not have an error handler. This caused the Flux to hang indefinitely instead of terminating with an error.

This commit adds an error handler to the inner subscriber that:
1. Cancels the upstream subscription to prevent further processing.
2. Propagates the error to the sink, allowing the caller to receive the error signal.
3. Updates the map operation to return the entity for better debugging capability.

Signed-off-by: Noel F <noel@Noels-MacBook-Pro.local>

* Add test for error propagation in reactive Flux save operations

This test verifies that errors occurring during saveAll operations
with a Flux are properly propagated to the subscriber instead of
being swallowed. The test creates a Flux that emits valid entities
followed by an error, and confirms the error reaches the caller.

Signed-off-by: Noel F <noel@Noels-MacBook-Pro.local>

* undo format fixes

Signed-off-by: Noel F <noel@Noels-MacBook-Pro.local>

* Update error propagation test: expect 0 entities before error due to race condition

The manual subscriber's onError fires before in-flight saveAll can push
results through tryEmitNext, so the caller sees 0 entities before the error.
Updated test expectation and added clarifying comment.

Signed-off-by: Noel F <noel@Noels-MacBook-Pro.local>

---------

Signed-off-by: Noel F <noel@Noels-MacBook-Pro.local>
Co-authored-by: xylos19 <noel@Noels-MacBook-Pro.local>

Closes #3233

Author: noel1155

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

@@ -245,7 +245,10 @@ public void onSubscribe(Subscription subscription) {
 						public void onNext(List<T> entityList) {
 							onNextHasBeenCalled.set(true);
 							saveAll(entityList, index)
-									.map(sink::tryEmitNext)
+									.map(entity -> {
+										sink.tryEmitNext(entity);
+										return entity;
+									})
 									.doOnComplete(() -> {
 										if (!upstreamComplete.get()) {
 											if (subscription == null) {
@@ -255,7 +258,14 @@ public void onNext(List<T> entityList) {
 										} else {
 											sink.tryEmitComplete();
 										}
-									}).subscribe();
+									})
+									.subscribe(v -> {
+									}, error -> {
+										if (subscription != null) {
+											subscription.cancel();
+										}
+										sink.tryEmitError(error);
+									});
 						}
 
 						@Override

src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchIntegrationTests.java

@@ -1220,6 +1220,28 @@ void shouldFailWithConflictOnAttemptToSaveWithSameVersion() {
 				.allMatch(failureStatus -> failureStatus.status().equals(409));
 	}
 
+	@Test // Error propagation in reactive Flux save
+	@DisplayName("should propagate errors during Flux save operations")
+	void shouldPropagateErrorsDuringFluxSaveOperations() {
+		// Create a Flux that will produce an error after emitting some valid entities
+		Flux<SampleEntity> entitiesWithError = Flux.concat(
+				Flux.just(
+						randomEntity("valid entity 1"),
+						randomEntity("valid entity 2")),
+				Flux.error(new RuntimeException("Simulated error during entity creation")));
+
+		// The save operation should propagate the error to the subscriber.
+		// With the manual subscriber approach, the error propagates eagerly —
+		// sink.tryEmitError is called before in-flight saveAll results can be emitted,
+		// so the caller sees 0 entities before the error.
+		operations.save(entitiesWithError, SampleEntity.class, 10)
+				.as(StepVerifier::create)
+				.expectNextCount(0)
+				.expectErrorMatches(throwable -> throwable instanceof RuntimeException &&
+						throwable.getMessage().equals("Simulated error during entity creation"))
+				.verify();
+	}
+
 	// endregion
 
 	// region Helper functions