Replace Ditto pubsub by direct shard messaging for search update.

Reason: Ditto pubsub reacts too slowly for coordinated shutdown. Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
eclipse-ditto · Aug 21, 2022 · 909ebea · 909ebea
1 parent 87ca5dc
commit 909ebea
Show file tree

Hide file tree

Showing 9 changed files with 37 additions and 228 deletions.
diff --git a/things/service/pom.xml b/things/service/pom.xml
@@ -96,6 +96,10 @@
             <groupId>org.eclipse.ditto</groupId>
             <artifactId>ditto-things-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.ditto</groupId>
+            <artifactId>ditto-thingsearch-api</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>org.eclipse.ditto</groupId>

diff --git a/...c/main/java/org/eclipse/ditto/things/service/persistence/actors/ThingSupervisorActor.java b/...c/main/java/org/eclipse/ditto/things/service/persistence/actors/ThingSupervisorActor.java
@@ -50,6 +50,7 @@
 import org.eclipse.ditto.things.service.common.config.DittoThingsConfig;
 import org.eclipse.ditto.things.service.enforcement.ThingEnforcement;
 import org.eclipse.ditto.things.service.enforcement.ThingEnforcerActor;
+import org.eclipse.ditto.thingsearch.api.ThingsSearchConstants;
 
 import akka.actor.ActorKilledException;
 import akka.actor.ActorRef;
@@ -137,9 +138,9 @@ private ThingSupervisorActor(final ActorRef pubSubMediator,
                 ThingsMessagingConstants.CLUSTER_ROLE,
                 ThingsMessagingConstants.SHARD_REGION
         );
-        // TODO
         searchShardRegionProxy =
-                shardRegionProxyActorFactory.getShardRegionProxyActor("search", "search-wildcard-updater");
+                shardRegionProxyActorFactory.getShardRegionProxyActor(ThingsSearchConstants.CLUSTER_ROLE,
+                        ThingsSearchConstants.SHARD_REGION);
 
         try {
             inlinePolicyEnrichment = new SupervisorInlinePolicyEnrichment(getContext().getSystem(), log, getEntityId(),

diff --git a/thingsearch/service/pom.xml b/thingsearch/service/pom.xml
@@ -71,10 +71,6 @@
             <groupId>org.eclipse.ditto</groupId>
             <artifactId>ditto-internal-utils-cache-loaders</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.eclipse.ditto</groupId>
-            <artifactId>ditto-internal-utils-pubsub-things</artifactId>
-        </dependency>
 
         <!-- logstash appender logging -->
         <dependency>

diff --git a/...src/main/java/org/eclipse/ditto/thingsearch/service/updater/actors/NewEventForwarder.java b/...src/main/java/org/eclipse/ditto/thingsearch/service/updater/actors/NewEventForwarder.java
diff --git a/...ava/org/eclipse/ditto/thingsearch/service/updater/actors/PolicyModificationForwarder.java b/...ava/org/eclipse/ditto/thingsearch/service/updater/actors/PolicyModificationForwarder.java
@@ -32,9 +32,11 @@
 import org.eclipse.ditto.thingsearch.service.common.config.DittoSearchConfig;
 import org.eclipse.ditto.thingsearch.service.persistence.write.ThingsSearchUpdaterPersistence;
 
+import akka.Done;
 import akka.NotUsed;
 import akka.actor.AbstractActor;
 import akka.actor.ActorRef;
+import akka.actor.CoordinatedShutdown;
 import akka.actor.Props;
 import akka.actor.Status;
 import akka.event.DiagnosticLoggingAdapter;
@@ -58,6 +60,7 @@ final class PolicyModificationForwarder extends AbstractActor {
 
     private final DiagnosticLoggingAdapter log = DittoLoggerFactory.getDiagnosticLoggingAdapter(this);
 
+    private final ActorRef pubSubMediator;
     private final ActorRef thingsUpdater;
     private final ThingsSearchUpdaterPersistence persistence;
     private final BlockNamespaceBehavior blockNamespaceBehavior;
@@ -72,6 +75,7 @@ private PolicyModificationForwarder(final ActorRef pubSubMediator,
             final BlockedNamespaces blockedNamespaces,
             final ThingsSearchUpdaterPersistence persistence) {
 
+        this.pubSubMediator = pubSubMediator;
         this.thingsUpdater = thingsUpdater;
         this.persistence = persistence;
         blockNamespaceBehavior = BlockNamespaceBehavior.of(blockedNamespaces);
@@ -97,13 +101,24 @@ public static Props props(final ActorRef pubSubMediator,
             final BlockedNamespaces blockedNamespaces,
             final ThingsSearchUpdaterPersistence persistence) {
 
-        return Props.create(PolicyModificationForwarder.class, pubSubMediator, thingsUpdater, blockedNamespaces, persistence);
+        return Props.create(PolicyModificationForwarder.class, pubSubMediator, thingsUpdater, blockedNamespaces,
+                persistence);
     }
 
     @Override
-    public void postStop() throws Exception {
+    public void preStart() {
+        CoordinatedShutdown.get(getContext().getSystem())
+                .addTask(CoordinatedShutdown.PhaseServiceUnbind(), "service-unbind-" + ACTOR_NAME, () -> {
+                    final var unsub = DistPubSubAccess.unsubscribeViaGroup(PolicyTag.PUB_SUB_TOPIC_MODIFIED,
+                            ACTOR_NAME, getSelf());
+                    final var shutdownAskTimeout = Duration.ofMinutes(1); // does not matter as phase will timeout
+                    return Patterns.ask(pubSubMediator, unsub, shutdownAskTimeout).thenApply(reply -> Done.done());
+                });
+    }
+
+    @Override
+    public void postStop() {
         terminateStream();
-        super.postStop();
     }
 
     @Override
@@ -163,7 +178,8 @@ private void dumpPolicyRevisions(final Control trigger) {
 
     private void streamTerminated(final Object streamTerminated) {
         if (streamTerminated instanceof Status.Failure failure) {
-            final String errorMessage = "PolicyModificationForwarder stream terminated (should NEVER happen!), restarting";
+            final String errorMessage =
+                    "PolicyModificationForwarder stream terminated (should NEVER happen!), restarting";
             log.error(failure.cause(), errorMessage);
         } else {
             log.info("PolicyModificationForwarder stream completed; restarting");

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
@@ -21,8 +21,6 @@
 import org.eclipse.ditto.internal.utils.health.RetrieveHealth;
 import org.eclipse.ditto.internal.utils.namespaces.BlockedNamespaces;
 import org.eclipse.ditto.internal.utils.persistence.mongo.DittoMongoClient;
-import org.eclipse.ditto.internal.utils.pubsub.DistributedAcks;
-import org.eclipse.ditto.internal.utils.pubsubthings.ThingEventPubSubFactory;
 import org.eclipse.ditto.thingsearch.api.ThingsSearchConstants;
 import org.eclipse.ditto.thingsearch.service.common.config.SearchConfig;
 import org.eclipse.ditto.thingsearch.service.common.util.RootSupervisorStrategyFactory;
@@ -110,16 +108,10 @@ private SearchUpdaterRootActor(final SearchConfig searchConfig,
 
         pubSubMediator.tell(DistPubSubAccess.put(getSelf()), getSelf());
 
-        final var thingEventSub =
-                ThingEventPubSubFactory.shardIdOnly(getContext(), numberOfShards, DistributedAcks.empty(actorSystem))
-                        .startDistributedSub();
         final var thingsUpdaterProps =
-                ThingsUpdater.props(thingEventSub, updaterShard, updaterConfig, blockedNamespaces,
-                        pubSubMediator);
+                ThingsUpdater.props(updaterShard, updaterConfig, blockedNamespaces, pubSubMediator);
 
         thingsUpdaterActor = startChildActor(ThingsUpdater.ACTOR_NAME, thingsUpdaterProps);
-        startClusterSingletonActor(NewEventForwarder.ACTOR_NAME,
-                NewEventForwarder.props(thingEventSub, updaterShard, blockedNamespaces));
 
         // start policy modification forwarder
         startChildActor(PolicyModificationForwarder.ACTOR_NAME, PolicyModificationForwarder.props(