FIX: Handling abnormal swithover from zk.

Author: brido4125

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

@@ -471,6 +471,17 @@ private void updateReplConnections(List<InetSocketAddress> addrs) throws IOExcep
       Set<ArcusReplNodeAddress> newSlaveAddrs = getSlaveAddrsFromGroupAddrs(newGroupAddrs);
 
       if (oldMasterAddr.isSameAddress(newMasterAddr)) {
+        if (oldGroup.getMasterCandidate() != null) {
+          /**
+           * Handling the below case.
+           * old group : [oldMaster, oldSlave]
+           * old group after switchover response : [oldMaster, oldSlave-masterCandidate]
+           * new group from zk cache list: [oldMaster, X]
+           */
+          taskList.add(new MoveOperationTask(oldGroup.getMasterCandidate(),
+                  oldMasterNode, false));
+          oldGroup.clearMasterCandidate();
+        }
         // add newly added slave node
         for (ArcusReplNodeAddress newSlaveAddr : newSlaveAddrs) {
           if (!oldSlaveAddrs.contains(newSlaveAddr)) {
@@ -490,8 +501,25 @@ private void updateReplConnections(List<InetSocketAddress> addrs) throws IOExcep
       } 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 */