handle SERVICE_REQUESTS_DONE inAnyState() of PersistenceCleanupActor;

unsubscribe from pubSub only if the subscription was done in AbstractPersistenceOperationsActor; Signed-off-by: Stefan Maute <stefan.maute@bosch.io>
eclipse-ditto · Sep 26, 2022 · 53178f2 · 53178f2
1 parent 2d1d870
commit 53178f2
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 15 deletions.
diff --git a/...lipse/ditto/internal/utils/persistence/operations/AbstractPersistenceOperationsActor.java b/...lipse/ditto/internal/utils/persistence/operations/AbstractPersistenceOperationsActor.java
@@ -353,20 +353,33 @@ private void serviceUnbind(final Control serviceUnbind) {
         logger.info("{}: unsubscribing from pubsub for {} actor", serviceUnbind, getActorName());
 
         final ActorRef self = getSelf();
-        final CompletableFuture<Done> unsubscribeTask = CompletableFuture.allOf(
-                        Patterns.ask(pubSubMediator,
-                                        DistPubSubAccess.unsubscribeViaGroup(PurgeEntities.getTopic(entityType),
-                                                getSubscribeGroup(), self),
-                                        SHUTDOWN_ASK_TIMEOUT)
-                                .toCompletableFuture(),
-                        Patterns.ask(pubSubMediator,
-                                        DistPubSubAccess.unsubscribeViaGroup(PurgeNamespace.TYPE, getSubscribeGroup(),
-                                                self), SHUTDOWN_ASK_TIMEOUT)
-                                .toCompletableFuture())
-                .thenApply(ack -> {
-                    logger.info("Unsubscribed successfully from pubsub for {} actor", getActorName());
-                    return Done.getInstance();
-                });
+        final CompletableFuture<Object> unsubscribeFromPurgeNamespace;
+        final CompletableFuture<Object> unsubscribeFromPurgeEntities;
+        if (null != namespaceOps) {
+            unsubscribeFromPurgeNamespace = Patterns.ask(pubSubMediator,
+                            DistPubSubAccess.unsubscribeViaGroup(PurgeNamespace.TYPE, getSubscribeGroup(), self),
+                            SHUTDOWN_ASK_TIMEOUT)
+                    .toCompletableFuture();
+        } else {
+            unsubscribeFromPurgeNamespace = CompletableFuture.completedFuture(Done.getInstance());
+        }
+
+        if (null != entitiesOps) {
+            unsubscribeFromPurgeEntities = Patterns.ask(pubSubMediator,
+                            DistPubSubAccess.unsubscribeViaGroup(PurgeEntities.getTopic(entityType), getSubscribeGroup(),
+                                    self),
+                            SHUTDOWN_ASK_TIMEOUT)
+                    .toCompletableFuture();
+        } else {
+            unsubscribeFromPurgeEntities = CompletableFuture.completedFuture(Done.getInstance());
+        }
+
+        final CompletableFuture<Done> unsubscribeTask =
+                CompletableFuture.allOf(unsubscribeFromPurgeNamespace, unsubscribeFromPurgeEntities)
+                        .thenApply(ack -> {
+                            logger.info("Unsubscribed successfully from pubsub for {} actor", getActorName());
+                            return Done.getInstance();
+                        });
 
         Patterns.pipe(unsubscribeTask, getContext().getDispatcher()).to(getSender());
     }

diff --git a/...va/org/eclipse/ditto/internal/utils/persistentactors/cleanup/PersistenceCleanupActor.java b/...va/org/eclipse/ditto/internal/utils/persistentactors/cleanup/PersistenceCleanupActor.java
@@ -77,7 +77,8 @@ public final class PersistenceCleanupActor extends AbstractFSM<PersistenceCleanu
      */
     private static final Duration SHUTDOWN_ASK_TIMEOUT = Duration.ofMinutes(2L);
 
-    private static final Throwable KILL_SWITCH_EXCEPTION = new IllegalStateException();
+    private static final Throwable KILL_SWITCH_EXCEPTION =
+            new IllegalStateException("Aborting stream because of graceful shutdown.");
 
     private final ThreadSafeDittoLoggingAdapter logger = DittoLoggerFactory.getThreadSafeDittoLoggingAdapter(this);
     private final Materializer materializer = Materializer.createMaterializer(getContext());
@@ -181,6 +182,7 @@ private FSMStateFunctionBuilder<State, String> inAnyState() {
                     final var stay = stay();
                     return setLastPid.map(stay::using).orElse(stay);
                 })
+                .eventEquals(Control.SERVICE_REQUESTS_DONE, this::serviceRequestsDone)
                 .anyEvent((message, lastPid) -> {
                     logger.warning("Got unhandled message <{}> when state=<{}> lastPid=<{}>",
                             message, stateName().name(), lastPid);