PersistenceCleanUpActor: Enable setting last-pid in ModifyConfig

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
eclipse-ditto · Sep 6, 2021 · f111946 · f111946
1 parent bff7b52
commit f111946
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 12 deletions.
diff --git a/...va/org/eclipse/ditto/internal/utils/persistentactors/cleanup/PersistenceCleanUpActor.java b/...va/org/eclipse/ditto/internal/utils/persistentactors/cleanup/PersistenceCleanUpActor.java
@@ -32,6 +32,7 @@
 import org.eclipse.ditto.internal.utils.metrics.instruments.counter.Counter;
 import org.eclipse.ditto.internal.utils.persistence.mongo.streaming.MongoReadJournal;
 import org.eclipse.ditto.json.JsonObject;
+import org.eclipse.ditto.json.JsonValue;
 
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
@@ -62,6 +63,13 @@ public final class PersistenceCleanUpActor extends AbstractFSM<PersistenceCleanU
      */
     public static final String NAME = "persistenceCleanUp";
 
+    /**
+     * JSON field of ModifyConfig commands that
+     */
+    private static final String SET_LAST_PID = "last-pid";
+
+    private static final Throwable KILL_SWITCH_EXCEPTION = new IllegalStateException();
+
     private final ThreadSafeDittoLoggingAdapter logger = DittoLoggerFactory.getThreadSafeDittoLoggingAdapter(this);
     private final Materializer materializer = Materializer.createMaterializer(getContext());
     private final Counter deleteEventsCounter = DittoMetrics.counter("cleanup_delete_events");
@@ -145,7 +153,12 @@ private FSMStateFunctionBuilder<State, String> inAnyState() {
                 })
                 .event(ModifyConfig.class, (modifyConfig, lastPid) -> {
                     modifyConfigBehavior().onMessage().apply(modifyConfig);
-                    return stay();
+                    final var setLastPid = modifyConfig.getConfig()
+                            .getValue(SET_LAST_PID)
+                            .filter(JsonValue::isString)
+                            .map(JsonValue::asString);
+                    final var stay = stay();
+                    return setLastPid.map(stay::using).orElse(stay);
                 })
                 .anyEvent((message, lastPid) -> {
                     logger.warning("Got unhandled message <{}> when state=<{}> lastPid=<{}>",
@@ -191,7 +204,7 @@ private FSM.State<State, String> streamComplete(final Control streamComplete, fi
         final var result = goTo(State.IN_QUIET_PERIOD).using("");
         if (config.isEnabled()) {
             final var nextQuietPeriod = randomizeQuietPeriod();
-            logger.info("Stream complete. Next stream in <{}>", nextQuietPeriod);
+            logger.info("Stream complete. Next stream in <{}> from start", nextQuietPeriod);
             return result.forMax(nextQuietPeriod);
         } else {
             logger.info("Stream complete and disabled.");
@@ -203,28 +216,29 @@ private FSM.State<State, String> streamFailed(final Control streamComplete, fina
         final var result = goTo(State.IN_QUIET_PERIOD).using(lastPid);
         if (config.isEnabled()) {
             final var nextQuietPeriod = randomizeQuietPeriod();
-            logger.info("Stream failed. Next stream in <{}> starting from <{}>", nextQuietPeriod, lastPid);
+            logger.info("Stream failed or shutdown. Next stream in <{}> starting from <{}>", nextQuietPeriod, lastPid);
             return result.forMax(nextQuietPeriod);
         } else {
-            logger.info("Stream failed and disabled.");
+            logger.info("Stream failed or shutdown and disabled. Last PID=<{}>", lastPid);
             return result;
         }
     }
 
     private FSM.State<State, String> shutdownRunningStream(final Control shutdown, final String lastPid) {
         logger.info("Activating kill-switch by demand: <{}>", killSwitch);
         if (killSwitch != null) {
-            killSwitch.shutdown();
+            // using ABORT to preserve lastPid
+            killSwitch.abort(KILL_SWITCH_EXCEPTION);
         }
         return stay();
     }
 
     private FSM.State<State, String> shutdownInQuietPeriod(final Control shutdown, final String lastPid) {
         if (config.isEnabled()) {
-            logger.info("Starting stream in <{}> on request", config.getQuietPeriod());
+            logger.info("Starting stream from <{}> in <{}> on request", lastPid, config.getQuietPeriod());
             return goTo(State.IN_QUIET_PERIOD).forMax(config.getQuietPeriod());
         } else {
-            logger.info("Stream disabled");
+            logger.info("Stream disabled. lastPid=<{}>", lastPid);
             return goTo(State.IN_QUIET_PERIOD);
         }
     }
@@ -257,7 +271,9 @@ private Done streamCompletedOrFailed(@Nullable final Done done, @Nullable final
         if (error == null) {
             getSelf().tell(Control.STREAM_COMPLETE, ActorRef.noSender());
         } else {
-            logger.error(error, "Stream failed");
+            if (error != KILL_SWITCH_EXCEPTION) {
+                logger.error(error, "Stream failed");
+            }
             getSelf().tell(Control.STREAM_FAILED, ActorRef.noSender());
         }
         return Done.getInstance();

diff --git a/...rg/eclipse/ditto/internal/utils/persistentactors/cleanup/PersistenceCleanUpActorTest.java b/...rg/eclipse/ditto/internal/utils/persistentactors/cleanup/PersistenceCleanUpActorTest.java
@@ -220,16 +220,19 @@ public void modifyConfigWhenRunning() {
             expectMsg(retrieveHealthResponse("RUNNING", ""));
 
             // WHEN config is modified
-            final var modifyConfig =
-                    ModifyConfig.of(JsonObject.newBuilder().set("enabled", false).build(), DittoHeaders.empty());
+            final var modifyConfig = ModifyConfig.of(JsonObject.newBuilder()
+                            .set("enabled", false)
+                            .set("last-pid", "thing:last:pid")
+                            .build(),
+                    DittoHeaders.empty());
             underTest.tell(modifyConfig, getRef());
             final var response = expectMsgClass(ModifyConfigResponse.class);
             final var expectedConfig = CleanUpConfig.of(ConfigFactory.parseMap(Map.of("clean-up.enabled", false)));
             assertThat(response.getConfig()).containsExactlyInAnyOrderElementsOf(
                     JsonObject.of(expectedConfig.render().root().render(ConfigRenderOptions.concise())));
 
             // THEN the running stream is shutdown
-            waitForResponse(this, underTest, retrieveHealthResponse("IN_QUIET_PERIOD", ""), () -> {
+            waitForResponse(this, underTest, retrieveHealthResponse("IN_QUIET_PERIOD", "thing:last:pid"), () -> {
                 try {
                     Thread.sleep(300L);
                 } catch (final Exception e) {
@@ -244,16 +247,19 @@ private void waitForResponse(final TestKit testKit,
             final RetrieveHealthResponse expectedResponse,
             final Runnable wait) {
         final var retries = 30;
+        Object lastResponse = null;
         for (int i = 1; i <= retries; ++i) {
             underTest.tell(RetrieveHealth.newInstance(), testKit.getRef());
             final var response = testKit.expectMsgClass(RetrieveHealthResponse.class);
             if (response.equals(expectedResponse)) {
                 return;
             }
+            lastResponse = response;
             Logging.getLogger(actorSystem, this).info("Waiting {}/{}", i, retries);
             wait.run();
         }
-        throw new AssertionError("Did not receive expected response: " + expectedResponse);
+        throw new AssertionError(
+                "Did not receive expected response: " + expectedResponse + "\nlastResponse=" + lastResponse);
     }
 
     private Props testProps() {