From 911d80e77e406907fc3134c307b5897fc265883f Mon Sep 17 00:00:00 2001 From: noel1155 <122672182+noel1155@users.noreply.github.com> Date: Sat, 14 Feb 2026 23:39:08 -0800 Subject: [PATCH] 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 * 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 * undo format fixes Signed-off-by: Noel F * 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 --------- Signed-off-by: Noel F Co-authored-by: xylos19 Closes #3233 (cherry picked from commit 0c1f5369dff05a60db43f81f048e776509b79330) --- ...AbstractReactiveElasticsearchTemplate.java | 14 ++++++++++-- ...ReactiveElasticsearchIntegrationTests.java | 22 +++++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java b/src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java index dabc838b6..c82e39a40 100644 --- a/src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java +++ b/src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java @@ -245,7 +245,10 @@ abstract public class AbstractReactiveElasticsearchTemplate public void onNext(List 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 @@ abstract public class AbstractReactiveElasticsearchTemplate } else { sink.tryEmitComplete(); } - }).subscribe(); + }) + .subscribe(v -> { + }, error -> { + if (subscription != null) { + subscription.cancel(); + } + sink.tryEmitError(error); + }); } @Override diff --git a/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchIntegrationTests.java b/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchIntegrationTests.java index 5495295cc..b274f90fe 100644 --- a/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchIntegrationTests.java +++ b/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchIntegrationTests.java @@ -1220,6 +1220,28 @@ public abstract class ReactiveElasticsearchIntegrationTests { .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 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