FIX: Handling abnormal swithover from zk.

Author: brido4125

src/main/java/net/spy/memcached/MemcachedConnection.java

@@ -485,13 +485,34 @@ private void updateReplConnections(List<InetSocketAddress> addrs) throws IOExcep
             // move operation slave -> master.
             taskList.add(new MoveOperationTask(
                 oldSlaveNode, oldMasterNode, false));
+            // clear the masterCandidate if the removed slave is the masterCandidate.
+            if (oldGroup.getMasterCandidate() == oldSlaveNode) {
+              oldGroup.clearMasterCandidate();
+            }
           }
         }
       } else if (oldSlaveAddrs.contains(newMasterAddr)) {
         if (newSlaveAddrs.contains(oldMasterAddr)) {
           // Switchover
-          if (oldGroup.getMasterCandidate() != null) {
+          MemcachedNode oldMasterCandidate = oldGroup.getMasterCandidate();
+          if (oldMasterCandidate != null) {
+            ArcusReplNodeAddress masterFromZk = (ArcusReplNodeAddress) oldGroup
+                    .getSlaveNodeBy(newMasterAddr.getIPPort()).getSocketAddress();
             changeRoleGroups.add(oldGroup);
+            if (!masterFromZk.isSameAddress(
+                    ((ArcusReplNodeAddress) oldMasterCandidate.getSocketAddress()))) {
+              /**
+               * Moves ops from oldMasterCandidate set by cache server to newMasterCandidate.
+               * Handling the below case.
+               * old group : [oldMaster, oldSlave1, oldSlave2]
+               * old group after switchover response :
+               * [oldMaster, oldSlave1-masterCandidate, oldSlave2]
+               * new group from zk cache list: [slave1, X, newMaster]
+               */
+              oldGroup.setMasterCandidateByAddr(newMasterAddr.getIPPort());
+              taskList.add(new MoveOperationTask(
+                      oldMasterCandidate, oldGroup.getMasterCandidate(), false));
+            }
           } else {
             // ZK event occurs before cache server response.
             oldGroup.setMasterCandidateByAddr(newMasterAddr.getIPPort());

src/main/java/net/spy/memcached/MemcachedReplicaGroup.java

@@ -86,6 +86,10 @@ public void setMasterCandidate() {
     }
   }
 
+  public void clearMasterCandidate() {
+    this.masterCandidate = null;
+  }
+
   public void setMasterCandidateByAddr(String address) {
     for (MemcachedNode node : this.getSlaveNodes()) {
       if (address.equals(((ArcusReplNodeAddress) node.getSocketAddress()).getIPPort())) {
@@ -181,5 +185,14 @@ private MemcachedNode getNextActiveSlaveNodeNoRotate() {
   public static String getGroupNameFromNode(final MemcachedNode node) {
     return ((ArcusReplNodeAddress) node.getSocketAddress()).getGroupName();
   }
+
+  public MemcachedNode getSlaveNodeBy(String address) {
+    for (MemcachedNode node : this.getSlaveNodes()) {
+      if (address.equals(((ArcusReplNodeAddress) node.getSocketAddress()).getIPPort())) {
+        return node;
+      }
+    }
+    return null;
+  }
 }
 /* ENABLE_REPLICATION end */