Add test for validating search consistency in case of conflicting sea…

…rch updaters Tests scenario where cluster sharding contains 2 updaters for same thing. (Can happen on failing cluster restarts) Signed-off-by: David Schwilk <david.schwilk@bosch.io>
eclipse-ditto · Mar 14, 2022 · 65a9e7a · 65a9e7a
1 parent ad0ef9f
commit 65a9e7a
Show file tree

Hide file tree

Showing 5 changed files with 564 additions and 6 deletions.
diff --git a/...rg/eclipse/ditto/thingsearch/service/persistence/write/streaming/SearchUpdaterStream.java b/...rg/eclipse/ditto/thingsearch/service/persistence/write/streaming/SearchUpdaterStream.java
@@ -27,9 +27,9 @@
 import com.mongodb.reactivestreams.client.MongoDatabase;
 
 import akka.NotUsed;
-import akka.actor.ActorContext;
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
+import akka.actor.ClassicActorSystemProvider;
 import akka.stream.Attributes;
 import akka.stream.KillSwitch;
 import akka.stream.KillSwitches;
@@ -113,16 +113,16 @@ public static SearchUpdaterStream of(final UpdaterConfig updaterConfig,
     /**
      * Start a perpetual search updater stream killed only by the kill-switch.
      *
-     * @param actorContext where to create actors for this stream.
+     * @param actorSystem where to create actors for this stream.
      * @return kill-switch to terminate the stream.
      */
-    public KillSwitch start(final ActorContext actorContext) {
+    public KillSwitch start(final ClassicActorSystemProvider actorSystem) {
         final Source<Source<AbstractWriteModel, NotUsed>, NotUsed> restartSource = createRestartSource();
         final Sink<Source<AbstractWriteModel, NotUsed>, NotUsed> restartSink = createRestartSink();
 
         return restartSource.viaMat(KillSwitches.single(), Keep.right())
                 .toMat(restartSink, Keep.left())
-                .run(actorContext.system());
+                .run(actorSystem);
     }
 
     private Source<Source<AbstractWriteModel, NotUsed>, NotUsed> createRestartSource() {

diff --git a/...ain/java/org/eclipse/ditto/thingsearch/service/updater/actors/SearchUpdaterRootActor.java b/...ain/java/org/eclipse/ditto/thingsearch/service/updater/actors/SearchUpdaterRootActor.java
@@ -105,7 +105,7 @@ private SearchUpdaterRootActor(final SearchConfig searchConfig,
                 SearchUpdaterStream.of(updaterConfig, actorSystem, thingsShard, policiesShard, updaterShard,
                         changeQueueActor, dittoMongoClient.getDefaultDatabase(), blockedNamespaces,
                         searchUpdateMapper);
-        updaterStreamKillSwitch = searchUpdaterStream.start(getContext());
+        updaterStreamKillSwitch = searchUpdaterStream.start(actorSystem);
 
         final var searchUpdaterPersistence =
                 MongoThingsSearchUpdaterPersistence.of(dittoMongoClient.getDefaultDatabase(),

diff --git a/...st/java/org/eclipse/ditto/thingsearch/service/updater/actors/BackgroundSyncActorTest.java b/...st/java/org/eclipse/ditto/thingsearch/service/updater/actors/BackgroundSyncActorTest.java
@@ -10,7 +10,6 @@
  *
  * SPDX-License-Identifier: EPL-2.0
  */
-
 package org.eclipse.ditto.thingsearch.service.updater.actors;
 
 import static org.assertj.core.api.Assertions.assertThat;