fixed bug that lead ConnectionActor to stop itself for brandnew conne…

…ctions

* and added ddata "remember-entities" in order to failover more gracefully in case of rolling update or failover

Signed-off-by: Thomas Jaeckle <thomas.jaeckle@bosch-si.com>

.gitignore

@@ -15,3 +15,4 @@ dependencies.txt
 /deployment/docker/sandbox/postgres
 /deployment/docker/sandbox/graphite/storage
 ajcore.*.txt
+*.mdb

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/ConnectionActor.java

@@ -96,6 +96,7 @@
 import akka.cluster.pubsub.DistributedPubSubMediator;
 import akka.cluster.routing.ClusterRouterPool;
 import akka.cluster.routing.ClusterRouterPoolSettings;
+import akka.cluster.sharding.ShardRegion;
 import akka.event.DiagnosticLoggingAdapter;
 import akka.japi.Creator;
 import akka.japi.pf.ReceiveBuilder;
@@ -123,9 +124,6 @@ public final class ConnectionActor extends AbstractPersistentActor {
     private static final String JOURNAL_PLUGIN_ID = "akka-contrib-mongodb-persistence-connection-journal";
     private static final String SNAPSHOT_PLUGIN_ID = "akka-contrib-mongodb-persistence-connection-snapshots";
 
-    private static final String UNRESOLVED_PLACEHOLDERS_MESSAGE =
-            "Failed to substitute all placeholders in '{}', target is dropped.";
-
     private static final String PUB_SUB_GROUP_PREFIX = "connection:";
 
     /**
@@ -281,8 +279,10 @@ public Receive createReceiveRecover() {
                             subscribeForEvents();
                         }
                         getContext().become(connectionCreatedBehaviour);
-                    } else {
+                    } else if (lastSequenceNr() > 0) {
+                        // if the last sequence number is already > 0 we can assume that the connection was deleted:
                         stopSelfIfDeletedAfterDelay();
+                        // otherwise not - as the connection may just be created!
                     }
 
                     getContext().getParent().tell(ConnectionSupervisorActor.ManualReset.getInstance(), getSelf());
@@ -911,9 +911,9 @@ private void stopChildActor(final ActorRef actor) {
     }
 
     private void stopSelf() {
-        log.debug("Shutting down");
-        // stop the supervisor (otherwise it'd restart this actor) which causes this actor to stop, too.
-        getContext().getParent().tell(PoisonPill.getInstance(), getSelf());
+        log.info("Passivating / shutting down");
+        final ShardRegion.Passivate passivateMessage = new ShardRegion.Passivate(PoisonPill.getInstance());
+        getContext().getParent().tell(passivateMessage, getSelf());
     }
 
     private void stopSelfIfDeletedAfterDelay() {

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/ConnectionSupervisorActor.java

@@ -34,6 +34,7 @@
 import akka.actor.Props;
 import akka.actor.SupervisorStrategy;
 import akka.actor.Terminated;
+import akka.cluster.sharding.ShardRegion;
 import akka.event.DiagnosticLoggingAdapter;
 import akka.japi.Creator;
 import akka.japi.pf.DeciderBuilder;
@@ -157,6 +158,7 @@ public Receive createReceive() {
                                     getContext().dispatcher(), null);
                     restartCount += 1;
                 })
+                .match(ShardRegion.Passivate.class, passivate -> getContext().getParent().tell(passivate, getSelf()))
                 .matchAny(message -> {
                     LogUtil.enhanceLogWithCustomField(log, BaseClientData.MDC_CONNECTION_ID, connectionId);
                     if (child != null) {

services/connectivity/starter/src/main/resources/connectivity.conf

@@ -115,6 +115,11 @@ akka {
   cluster {
     sharding {
       role = "connectivity"
+
+      # When this is set to 'on' the active entity actors will automatically be restarted
+      # upon Shard restart. i.e. if the Shard is started on a different ShardRegion
+      # due to rebalance or crash.
+      remember-entities = on
     }
 
     roles = [