X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-clustering-commons%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fcommon%2Factor%2FQuarantinedMonitorActor.java;h=c5c19d8d37ebf81300e7021d84c1784064d12116;hb=HEAD;hp=52df6ab7b388314879e72160890a624fc0104529;hpb=0f02b7edeb1454c1a568f0f1b050757e7503ddf7;p=controller.git

diff --git a/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/common/actor/QuarantinedMonitorActor.java b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/common/actor/QuarantinedMonitorActor.java
index 52df6ab7b3..c5c19d8d37 100644
--- a/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/common/actor/QuarantinedMonitorActor.java
+++ b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/common/actor/QuarantinedMonitorActor.java
@@ -8,10 +8,18 @@
 
 package org.opendaylight.controller.cluster.common.actor;
 
+import akka.actor.Address;
 import akka.actor.Props;
 import akka.actor.UntypedAbstractActor;
+import akka.cluster.Cluster;
+import akka.cluster.ClusterEvent;
 import akka.japi.Effect;
-import akka.remote.ThisActorSystemQuarantinedEvent;
+import akka.remote.AssociationErrorEvent;
+import akka.remote.RemotingLifecycleEvent;
+import akka.remote.artery.ThisActorSystemQuarantinedEvent;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.HashSet;
+import java.util.Set;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -24,20 +32,25 @@ import org.slf4j.LoggerFactory;
  *
  */
 public class QuarantinedMonitorActor extends UntypedAbstractActor {
+    public static final String ADDRESS = "quarantined-monitor";
 
     private static final Logger LOG = LoggerFactory.getLogger(QuarantinedMonitorActor.class);
-
-    public static final String ADDRESS = "quarantined-monitor";
+    private static final Integer MESSAGE_THRESHOLD = 10;
 
     private final Effect callback;
     private boolean quarantined;
 
+    private final Set<Address> addressSet = new HashSet<>();
+    private int count = 0;
+
+    @SuppressFBWarnings(value = "MC_OVERRIDABLE_METHOD_CALL_IN_CONSTRUCTOR", justification = "Akka class design")
     protected QuarantinedMonitorActor(final Effect callback) {
         this.callback = callback;
 
         LOG.debug("Created QuarantinedMonitorActor");
 
-        getContext().system().eventStream().subscribe(getSelf(), ThisActorSystemQuarantinedEvent.class);
+        getContext().system().eventStream().subscribe(getSelf(), RemotingLifecycleEvent.class);
+        getContext().system().eventStream().subscribe(getSelf(), ClusterEvent.MemberDowned.class);
     }
 
     @Override
@@ -55,13 +68,39 @@ public class QuarantinedMonitorActor extends UntypedAbstractActor {
             return;
         }
 
-        if (message instanceof ThisActorSystemQuarantinedEvent) {
-            final ThisActorSystemQuarantinedEvent event = (ThisActorSystemQuarantinedEvent) message;
+        if (message instanceof ThisActorSystemQuarantinedEvent event) {
             LOG.warn("Got quarantined by {}", event.remoteAddress());
             quarantined = true;
 
             // execute the callback
             callback.apply();
+        } else if (message instanceof AssociationErrorEvent event) {
+            final String errorMessage = message.toString();
+            LOG.trace("errorMessage:{}", errorMessage);
+            if (errorMessage.contains("The remote system has a UID that has been quarantined")) {
+                final Address address = event.getRemoteAddress();
+                addressSet.add(address);
+                count++;
+                LOG.trace("address:{} addressSet: {} count:{}", address, addressSet, count);
+                if (count >= MESSAGE_THRESHOLD && addressSet.size() > 1) {
+                    count = 0;
+                    addressSet.clear();
+                    LOG.warn("Got quarantined via AssociationEvent by {}", event.remoteAddress());
+                    quarantined = true;
+
+                    // execute the callback
+                    callback.apply();
+                }
+            } else if (errorMessage.contains("The remote system explicitly disassociated")) {
+                count = 0;
+                addressSet.clear();
+            }
+        } else if (message instanceof ClusterEvent.MemberDowned event) {
+            if (Cluster.get(getContext().system()).selfMember().equals(event.member())) {
+                LOG.warn("This member has been downed, restarting");
+
+                callback.apply();
+            }
         }
     }