From 6b95b4a8a4a49dc7877271118c36d5e916d336ab Mon Sep 17 00:00:00 2001
From: Ted Yu <tedyu@apache.org>
Date: Fri, 3 Oct 2014 21:31:50 +0000
Subject: [PATCH] HBASE-12136 Race condition between client adding tableCF
 replication znode and  server triggering TableCFsTracker (Virag Kothari)

---
 .../replication/ReplicationPeerZKImpl.java    |  9 +++++++-
 .../replication/ReplicationPeersZKImpl.java   | 23 +++++++++++--------
 2 files changed, 22 insertions(+), 10 deletions(-)

diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerZKImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerZKImpl.java
index 13d9df3013c..848303cbb5f 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerZKImpl.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerZKImpl.java
@@ -261,12 +261,19 @@ public class ReplicationPeerZKImpl implements ReplicationPeer, Abortable, Closea
         Abortable abortable) {
       super(watcher, tableCFsZNode, abortable);
     }
+    
+    @Override
+    public synchronized void nodeCreated(String path) {
+      if (path.equals(node)) {
+        super.nodeCreated(path);
+        readTableCFsZnode();
+      }
+    }
 
     @Override
     public synchronized void nodeDataChanged(String path) {
       if (path.equals(node)) {
         super.nodeDataChanged(path);
-        readTableCFsZnode();
       }
     }
   }
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeersZKImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeersZKImpl.java
index dce090394d2..abb0984ef67 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeersZKImpl.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeersZKImpl.java
@@ -19,6 +19,7 @@
 package org.apache.hadoop.hbase.replication;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -42,6 +43,7 @@ import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.hadoop.hbase.zookeeper.ZKUtil.ZKUtilOp;
 import org.apache.zookeeper.KeeperException;
 
 import com.google.protobuf.ByteString;
@@ -109,18 +111,21 @@ public class ReplicationPeersZKImpl extends ReplicationStateZKBase implements Re
             + " because that id already exists.");
       }
       ZKUtil.createWithParents(this.zookeeper, this.peersZNode);
-      ZKUtil.createAndWatch(this.zookeeper, ZKUtil.joinZNode(this.peersZNode, id),
+      List<ZKUtilOp> listOfOps = new ArrayList<ZKUtil.ZKUtilOp>();
+      ZKUtilOp op1 = ZKUtilOp.createAndFailSilent(ZKUtil.joinZNode(this.peersZNode, id),
         toByteArray(peerConfig));
-      // There is a race b/w PeerWatcher and ReplicationZookeeper#add method to create the
-      // peer-state znode. This happens while adding a peer.
+      // There is a race (if hbase.zookeeper.useMulti is false) 
+      // b/w PeerWatcher and ReplicationZookeeper#add method to create the
+      // peer-state znode. This happens while adding a peer
       // The peer state data is set as "ENABLED" by default.
-      ZKUtil.createNodeIfNotExistsAndWatch(this.zookeeper, getPeerStateNode(id),
-        ENABLED_ZNODE_BYTES);
-      // A peer is enabled by default
-
+      ZKUtilOp op2 = ZKUtilOp.createAndFailSilent(getPeerStateNode(id), ENABLED_ZNODE_BYTES);
       String tableCFsStr = (tableCFs == null) ? "" : tableCFs;
-      ZKUtil.createNodeIfNotExistsAndWatch(this.zookeeper, getTableCFsNode(id),
-                    Bytes.toBytes(tableCFsStr));
+      ZKUtilOp op3 = ZKUtilOp.createAndFailSilent(getTableCFsNode(id), Bytes.toBytes(tableCFsStr));
+      listOfOps.add(op1);
+      listOfOps.add(op2);
+      listOfOps.add(op3);
+      ZKUtil.multiOrSequential(this.zookeeper, listOfOps, false);
+      // A peer is enabled by default
     } catch (KeeperException e) {
       throw new ReplicationException("Could not add peer with id=" + id
           + ", peerConfif=>" + peerConfig, e);