HBASE-15958 Implement ClaimQueues on top of HBase

Building on HBase-15883.
Now implementing the claim queues procedure within an HBase table.
Also added UnitTests to test claimQueue.
Peer tracking will still be performed by ZooKeeper though.
Also modified the queueId tracking procedure so we no longer have to perform scans over the Replication Table.
This does make our queue naming schema slightly different from ReplicationQueuesZKImpl though.

Signed-off-by: Elliott Clark <eclark@apache.org>

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

@@ -19,8 +19,8 @@
 package org.apache.hadoop.hbase.replication;
 
 import java.util.List;
-import java.util.SortedMap;
-import java.util.SortedSet;
+import java.util.Map;
+import java.util.Set;
 
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
 
@@ -96,10 +96,10 @@ public interface ReplicationQueues {
   /**
    * Take ownership for the set of queues belonging to a dead region server.
    * @param regionserver the id of the dead region server
-   * @return A SortedMap of the queues that have been claimed, including a SortedSet of WALs in
+   * @return A Map of the queues that have been claimed, including a Set of WALs in
    *         each queue. Returns an empty map if no queues were failed-over.
    */
-  SortedMap<String, SortedSet<String>> claimQueues(String regionserver);
+  Map<String, Set<String>> claimQueues(String regionserver);
 
   /**
    * Get a list of all region servers that have outstanding replication queues. These servers could

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

@@ -56,11 +56,11 @@ public void setConf(Configuration conf) {
     this.conf = conf;
   }
 
-  public Abortable getAbort() {
+  public Abortable getAbortable() {
     return abort;
   }
 
-  public void setAbort(Abortable abort) {
+  public void setAbortable(Abortable abort) {
     this.abort = abort;
   }
 }

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

@@ -19,11 +19,11 @@
 package org.apache.hadoop.hbase.replication;
 
 import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
-import java.util.SortedMap;
-import java.util.SortedSet;
-import java.util.TreeMap;
-import java.util.TreeSet;
+import java.util.Map;
+import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -73,7 +73,7 @@ public class ReplicationQueuesZKImpl extends ReplicationStateZKBase implements R
   private static final Log LOG = LogFactory.getLog(ReplicationQueuesZKImpl.class);
 
   public ReplicationQueuesZKImpl(ReplicationQueuesArguments args) {
-    this(args.getZk(), args.getConf(), args.getAbort());
+    this(args.getZk(), args.getConf(), args.getAbortable());
   }
 
   public ReplicationQueuesZKImpl(final ZooKeeperWatcher zk, Configuration conf,
@@ -178,8 +178,8 @@ public boolean isThisOurRegionServer(String regionserver) {
   }
 
   @Override
-  public SortedMap<String, SortedSet<String>> claimQueues(String regionserverZnode) {
-    SortedMap<String, SortedSet<String>> newQueues = new TreeMap<String, SortedSet<String>>();
+  public Map<String, Set<String>> claimQueues(String regionserverZnode) {
+    Map<String, Set<String>> newQueues = new HashMap<>();
     // check whether there is multi support. If yes, use it.
     if (conf.getBoolean(HConstants.ZOOKEEPER_USEMULTI, true)) {
       LOG.info("Atomically moving " + regionserverZnode + "'s WALs to my queue");
@@ -304,8 +304,8 @@ private void deleteAnotherRSQueues(String regionserverZnode) {
    * @param znode pertaining to the region server to copy the queues from
    * @return WAL queues sorted per peer cluster
    */
-  private SortedMap<String, SortedSet<String>> copyQueuesFromRSUsingMulti(String znode) {
-    SortedMap<String, SortedSet<String>> queues = new TreeMap<String, SortedSet<String>>();
+  private Map<String, Set<String>> copyQueuesFromRSUsingMulti(String znode) {
+    Map<String, Set<String>> queues = new HashMap<>();
     // hbase/replication/rs/deadrs
     String deadRSZnodePath = ZKUtil.joinZNode(this.queuesZNode, znode);
     List<String> peerIdsToProcess = null;
@@ -330,7 +330,7 @@ private SortedMap<String, SortedSet<String>> copyQueuesFromRSUsingMulti(String z
           continue; // empty log queue.
         }
         // create the new cluster znode
-        SortedSet<String> logQueue = new TreeSet<String>();
+        Set<String> logQueue = new HashSet<String>();
         queues.put(newPeerId, logQueue);
         ZKUtilOp op = ZKUtilOp.createAndFailSilent(newPeerZnode, HConstants.EMPTY_BYTE_ARRAY);
         listOfOps.add(op);
@@ -373,10 +373,10 @@ private SortedMap<String, SortedSet<String>> copyQueuesFromRSUsingMulti(String z
    * @param znode server names to copy
    * @return all wals for all peers of that cluster, null if an error occurred
    */
-  private SortedMap<String, SortedSet<String>> copyQueuesFromRS(String znode) {
+  private Map<String, Set<String>> copyQueuesFromRS(String znode) {
     // TODO this method isn't atomic enough, we could start copying and then
     // TODO fail for some reason and we would end up with znodes we don't want.
-    SortedMap<String, SortedSet<String>> queues = new TreeMap<String, SortedSet<String>>();
+    Map<String, Set<String>> queues = new HashMap<>();
     try {
       String nodePath = ZKUtil.joinZNode(this.queuesZNode, znode);
       List<String> clusters = ZKUtil.listChildrenNoWatch(this.zookeeper, nodePath);
@@ -406,7 +406,7 @@ private SortedMap<String, SortedSet<String>> copyQueuesFromRS(String znode) {
         }
         ZKUtil.createNodeIfNotExistsAndWatch(this.zookeeper, newClusterZnode,
           HConstants.EMPTY_BYTE_ARRAY);
-        SortedSet<String> logQueue = new TreeSet<String>();
+        Set<String> logQueue = new HashSet<String>();
         queues.put(newCluster, logQueue);
         for (String wal : wals) {
           String z = ZKUtil.joinZNode(clusterPath, wal);

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java

@@ -31,7 +31,6 @@
 import java.util.Map;
 import java.util.Random;
 import java.util.Set;
-import java.util.SortedMap;
 import java.util.SortedSet;
 import java.util.TreeSet;
 import java.util.UUID;
@@ -642,7 +641,7 @@ public void run() {
         LOG.info("Not transferring queue since we are shutting down");
         return;
       }
-      SortedMap<String, SortedSet<String>> newQueues = null;
+      Map<String, Set<String>> newQueues = null;
 
       newQueues = this.rq.claimQueues(rsZnode);
 
@@ -653,9 +652,9 @@ public void run() {
         return;
       }
 
-      for (Map.Entry<String, SortedSet<String>> entry : newQueues.entrySet()) {
+      for (Map.Entry<String, Set<String>> entry : newQueues.entrySet()) {
         String peerId = entry.getKey();
-        SortedSet<String> walsSet = entry.getValue();
+        Set<String> walsSet = entry.getValue();
         try {
           // there is not an actual peer defined corresponding to peerId for the failover.
           ReplicationQueueInfo replicationQueueInfo = new ReplicationQueueInfo(peerId);

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

@@ -22,8 +22,8 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.SortedMap;
-import java.util.SortedSet;
+import java.util.Map;
+import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -146,7 +146,7 @@ public void testReplicationQueues() throws ReplicationException {
     assertEquals(0, rq3.claimQueues(server1).size());
     assertEquals(2, rq3.getListOfReplicators().size());
 
-    SortedMap<String, SortedSet<String>> queues = rq2.claimQueues(server3);
+    Map<String, Set<String>> queues = rq2.claimQueues(server3);
     assertEquals(5, queues.size());
     assertEquals(1, rq2.getListOfReplicators().size());