HBASE-11394 Replication can have data loss if peer id contains hyphen "-"

hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeersZKImpl.java

@@ -110,6 +110,11 @@ public class ReplicationPeersZKImpl extends ReplicationStateZKBase implements Re
         throw new IllegalArgumentException("Cannot add a peer with id=" + id
             + " because that id already exists.");
       }
+      
+      if(id.contains("-")){
+        throw new IllegalArgumentException("Found invalid peer name:" + id);
+      }
+      
       ZKUtil.createWithParents(this.zookeeper, this.peersZNode);
       List<ZKUtilOp> listOfOps = new ArrayList<ZKUtil.ZKUtilOp>();
       ZKUtilOp op1 = ZKUtilOp.createAndFailSilent(ZKUtil.joinZNode(this.peersZNode, id),

hbase-client/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java

@@ -1316,7 +1316,11 @@ public class ZKUtil {
           deleteNodeRecursively(zkw, joinZNode(node, child));
         }
       }
+      //Zookeeper Watches are one time triggers; When children of parent nodes are deleted recursively. 
+      //Must set another watch, get notified of delete node   
+      if (zkw.getRecoverableZooKeeper().exists(node, zkw) != null){
         zkw.getRecoverableZooKeeper().delete(node, -1);
+      }
     } catch(InterruptedException ie) {
       zkw.interruptedException(ie);
     }

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationTrackerZKImpl.java

@@ -143,7 +143,7 @@ public class TestReplicationTrackerZKImpl {
     assertEquals("hostname2.example.org:1234", rsRemovedData);
   }
 
-  @Ignore ("Flakey") @Test(timeout = 30000)
+  @Test(timeout = 30000)
   public void testPeerRemovedEvent() throws Exception {
     rp.addPeer("5", new ReplicationPeerConfig().setClusterKey(utility.getClusterKey()), null);
     rt.registerListener(new DummyReplicationListener());
@@ -155,7 +155,7 @@ public class TestReplicationTrackerZKImpl {
     assertEquals("5", peerRemovedData);
   }
 
-  @Ignore ("Flakey") @Test(timeout = 30000)
+  @Test(timeout = 30000)
   public void testPeerListChangedEvent() throws Exception {
     // add a peer
     rp.addPeer("5", new ReplicationPeerConfig().setClusterKey(utility.getClusterKey()), null);
@@ -172,7 +172,32 @@ public class TestReplicationTrackerZKImpl {
     assertTrue(plChangedData.contains("5"));
 
     // clean up
-    ZKUtil.deleteNode(zkw, "/hbase/replication/peers/5");
+    //ZKUtil.deleteNode(zkw, "/hbase/replication/peers/5");
+    rp.removePeer("5");
+  }
+
+  @Test(timeout = 30000)
+  public void testPeerNameControl() throws Exception {
+    int exists = 0;
+    int hyphen = 0;
+    rp.addPeer("6", new ReplicationPeerConfig().setClusterKey(utility.getClusterKey()), null);
+    
+    try{
+      rp.addPeer("6", new ReplicationPeerConfig().setClusterKey(utility.getClusterKey()), null);
+    }catch(IllegalArgumentException e){
+      exists++;
+    }
+
+    try{
+      rp.addPeer("6-ec2", new ReplicationPeerConfig().setClusterKey(utility.getClusterKey()), null);
+    }catch(IllegalArgumentException e){
+      hyphen++;
+    }
+    assertEquals(1, exists);
+    assertEquals(1, hyphen);
+    
+    // clean up
+    rp.removePeer("6");
   }
   
   private class DummyReplicationListener implements ReplicationListener {