Handle empty update result in ThingUpdater.

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
eclipse-ditto · Apr 7, 2022 · a5ba29a · a5ba29a
1 parent 00618b8
commit a5ba29a
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 7 deletions.
diff --git a/...eclipse/ditto/thingsearch/service/persistence/write/streaming/BulkWriteResultAckFlow.java b/...eclipse/ditto/thingsearch/service/persistence/write/streaming/BulkWriteResultAckFlow.java
@@ -66,8 +66,7 @@ public static Pair<Status, List<String>> checkBulkWriteResult(final WriteResultA
 
         if (wasNotAcknowledged(writeResultAndErrors)) {
             // All failed.
-            acknowledgeFailures(getAllMetadata(writeResultAndErrors)
-            );
+            acknowledgeFailures(getAllMetadata(writeResultAndErrors));
             return Pair.create(Status.UNACKNOWLEDGED,
                     List.of(logResult("NotAcknowledged", writeResultAndErrors, false, false)));
         } else {

diff --git a/...vice/src/main/java/org/eclipse/ditto/thingsearch/service/updater/actors/ThingUpdater.java b/...vice/src/main/java/org/eclipse/ditto/thingsearch/service/updater/actors/ThingUpdater.java
@@ -53,6 +53,7 @@
 
 import com.mongodb.client.model.DeleteOneModel;
 
+import akka.Done;
 import akka.NotUsed;
 import akka.actor.AbstractFSMWithStash;
 import akka.actor.ActorRef;
@@ -213,6 +214,7 @@ private FSMStateFunctionBuilder<State, Data> ready() {
 
     private FSMStateFunctionBuilder<State, Data> persisting() {
         return matchEvent(Result.class, this::onResult)
+                .event(Done.class, this::onDone)
                 .event(ThingEvent.class, this::onEventWhenPersisting)
                 .event(ReceiveTimeout.class, this::shutdown)
                 .event(SHUTDOWN_CLASS, this::shutdownNow)
@@ -285,22 +287,32 @@ private FSM.State<State, Data> onResult(final Result result, final Data data) {
         };
     }
 
+    private FSM.State<State, Data> onDone(final Done done, final Data data) {
+        killSwitch = null;
+        log.debug("Update skipped");
+        return goTo(State.READY);
+    }
+
     private FSM.State<State, Data> tick(final Control tick, final Data data) {
         if (shouldPersist(data.metadata(), data.lastWriteModel().getMetadata())) {
             final var pair = Source.single(data)
                     .viaMat(KillSwitches.single(), Keep.right())
                     .via(flow)
-                    .toMat(Sink.head(), Keep.both())
+                    .toMat(Sink.seq(), Keep.both())
                     .run(materializer);
 
             killSwitch = pair.first();
             final var resultFuture = pair.second().handle((result, error) -> {
-                if (error != null || result == null) {
-                    final var errorToReport =
-                            error != null ? error : new IllegalStateException("Persistence produced no result");
+                if (error != null || result == null || result.size() > 1) {
+                    final var errorToReport = error != null
+                            ? error
+                            : new IllegalStateException("Got unexpected persistence results: " + result);
                     return Result.fromError(data.metadata(), errorToReport);
+                } else if (result.isEmpty()) {
+                    // no result; no change sent to persistence and all acknowledgement requests were fulfilled.
+                    return Done.done();
                 } else {
-                    return result;
+                    return result.get(0);
                 }
             });