diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeer.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeer.java index 936f89b..81efa28 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeer.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeer.java @@ -367,12 +367,19 @@ public class ReplicationPeer implements Abortable, Closeable { 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 df0e385..ceb972a 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 @@ -41,6 +41,7 @@ import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.zookeeper.ZKClusterId; import org.apache.hadoop.hbase.zookeeper.ZKUtil; +import org.apache.hadoop.hbase.zookeeper.ZKUtil.ZKUtilOp; import org.apache.hadoop.hbase.zookeeper.ZooKeeperListener; import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher; import org.apache.zookeeper.KeeperException; @@ -116,21 +117,24 @@ 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), - toByteArray(clusterKey)); - // There is a race b/w PeerWatcher and ReplicationZookeeper#add method to create the - // peer-state znode. This happens while adding a peer. + List listOfOps = new ArrayList(); + ZKUtilOp op1 = + ZKUtilOp.createAndFailSilent(ZKUtil.joinZNode(this.peersZNode, id), + toByteArray(clusterKey)); + // 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); } catch (KeeperException e) { - throw new ReplicationException("Could not add peer with id=" + id - + ", clusterKey=" + clusterKey, e); + throw new ReplicationException("Could not add peer with id=" + id + ", clusterKey=" + + clusterKey, e); } }