HBASE-27109 Move replication queue storage from zookeeper to a separated HBase table

hbase-client/src/main/java/org/apache/hadoop/hbase/util/JsonMapper.java

@@ -40,4 +40,8 @@ public static String writeMapAsString(Map<String, Object> map) throws IOExceptio
   public static String writeObjectAsString(Object object) throws IOException {
     return GSON.toJson(object);
   }
+
+  public static <T> T fromJson(String json, Class<T> clazz) {
+    return GSON.fromJson(json, clazz);
+  }
 }

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

@@ -220,7 +220,11 @@ public String getRsPath(ServerName sn) {
    * @param suffix ending of znode name
    * @return result of properly joining prefix with suffix
    */
-  public static String joinZNode(String prefix, String suffix) {
-    return prefix + ZNodePaths.ZNODE_PATH_SEPARATOR + suffix;
+  public static String joinZNode(String prefix, String... suffix) {
+    StringBuilder sb = new StringBuilder(prefix);
+    for (String s : suffix) {
+      sb.append(ZNodePaths.ZNODE_PATH_SEPARATOR).append(s);
+    }
+    return sb.toString();
   }
 }

hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/LockedResourceType.java

@@ -26,5 +26,6 @@ public enum LockedResourceType {
   TABLE,
   REGION,
   PEER,
-  META
+  META,
+  GLOBAL
 }

hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/Procedure.java

@@ -21,6 +21,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
 import org.apache.hadoop.hbase.exceptions.TimeoutIOException;
 import org.apache.hadoop.hbase.metrics.Counter;
 import org.apache.hadoop.hbase.metrics.Histogram;
@@ -33,6 +34,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos.ProcedureState;
 
 /**
@@ -1011,6 +1013,19 @@ final void doReleaseLock(TEnvironment env, ProcedureStore store) {
     releaseLock(env);
   }
 
+  protected final ProcedureSuspendedException suspend(int timeoutMillis, boolean jitter)
+    throws ProcedureSuspendedException {
+    if (jitter) {
+      // 10% possible jitter
+      double add = (double) timeoutMillis * ThreadLocalRandom.current().nextDouble(0.1);
+      timeoutMillis += add;
+    }
+    setTimeout(timeoutMillis);
+    setState(ProcedureProtos.ProcedureState.WAITING_TIMEOUT);
+    skipPersistence();
+    throw new ProcedureSuspendedException();
+  }
+
   @Override
   public int compareTo(final Procedure<TEnvironment> other) {
     return Long.compare(getProcId(), other.getProcId());

hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/TimeoutExecutorThread.java

@@ -78,9 +78,13 @@ public void add(InlineChore chore) {
   }
 
   public void add(Procedure<TEnvironment> procedure) {
-    LOG.info("ADDED {}; timeout={}, timestamp={}", procedure, procedure.getTimeout(),
-      procedure.getTimeoutTimestamp());
-    queue.add(new DelayedProcedure<>(procedure));
+    if (procedure.getTimeout() > 0) {
+      LOG.info("ADDED {}; timeout={}, timestamp={}", procedure, procedure.getTimeout(),
+        procedure.getTimeoutTimestamp());
+      queue.add(new DelayedProcedure<>(procedure));
+    } else {
+      LOG.info("Got negative timeout {} for {}, skip adding", procedure.getTimeout(), procedure);
+    }
   }
 
   public boolean remove(Procedure<TEnvironment> procedure) {

hbase-protocol-shaded/src/main/protobuf/server/master/MasterProcedure.proto

@@ -515,6 +515,7 @@ message UpdatePeerConfigStateData {
 
 message RemovePeerStateData {
   optional ReplicationPeer peer_config = 1;
+  repeated int64 ongoing_assign_replication_queues_proc_ids = 2;
 }
 
 message EnablePeerStateData {
@@ -679,16 +680,13 @@ message ClaimReplicationQueueRemoteStateData {
   required ServerName crashed_server = 1;
   required string queue = 2;
   required ServerName target_server = 3;
+  optional ServerName source_server = 4;
 }
 
 message ClaimReplicationQueueRemoteParameter {
   required ServerName crashed_server = 1;
   required string queue = 2;
-}
-
-enum ClaimReplicationQueuesState {
-  CLAIM_REPLICATION_QUEUES_DISPATCH = 1;
-  CLAIM_REPLICATION_QUEUES_FINISH = 2;
+  optional ServerName source_server = 3;
 }
 
 enum ModifyTableDescriptorState {
@@ -715,3 +713,27 @@ message ModifyStoreFileTrackerStateData {
 message ModifyColumnFamilyStoreFileTrackerStateData {
   required bytes family = 1;
 }
+
+enum AssignReplicationQueuesState {
+  ASSIGN_REPLICATION_QUEUES_ADD_MISSING_QUEUES = 1;
+  ASSIGN_REPLICATION_QUEUES_CLAIM = 2;
+  ASSIGN_REPLICATION_QUEUES_REMOVE_QUEUES = 3;
+}
+
+message AssignReplicationQueuesStateData {
+  required ServerName crashed_server = 1;
+}
+
+enum MigrateReplicationQueueFromZkToTableState {
+  MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_DISABLE_CLEANER = 1;
+  MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_PREPARE = 2;
+  MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_DISABLE_PEER = 3;
+  MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_MIGRATE = 4;
+  MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_WAIT_UPGRADING = 5;
+  MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_ENABLE_PEER = 6;
+  MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_ENABLE_CLEANER = 7;
+}
+
+message MigrateReplicationQueueFromZkToTableStateData {
+  repeated string disabled_peer_id = 1;
+}

hbase-replication/pom.xml

@@ -104,6 +104,16 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-library</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>

hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationGroupOffset.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.replication;
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+public class ReplicationGroupOffset {
+
+  public static final ReplicationGroupOffset BEGIN = new ReplicationGroupOffset("", 0L);
+
+  private final String wal;
+
+  private final long offset;
+
+  public ReplicationGroupOffset(String wal, long offset) {
+    this.wal = wal;
+    this.offset = offset;
+  }
+
+  public String getWal() {
+    return wal;
+  }
+
+  /**
+   * A negative value means this file has already been fully replicated out
+   */
+  public long getOffset() {
+    return offset;
+  }
+
+  @Override
+  public String toString() {
+    return wal + ":" + offset;
+  }
+
+  public static ReplicationGroupOffset parse(String str) {
+    int index = str.lastIndexOf(':');
+    return new ReplicationGroupOffset(str.substring(0, index),
+      Long.parseLong(str.substring(index + 1)));
+  }
+}

hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationQueueData.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.replication;
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap;
+
+/**
+ * Representing all the information for a replication queue.
+ */
+@InterfaceAudience.Private
+public class ReplicationQueueData {
+
+  private final ReplicationQueueId id;
+
+  private final ImmutableMap<String, ReplicationGroupOffset> offsets;
+
+  public ReplicationQueueData(ReplicationQueueId id,
+    ImmutableMap<String, ReplicationGroupOffset> offsets) {
+    this.id = id;
+    this.offsets = offsets;
+  }
+
+  public ReplicationQueueId getId() {
+    return id;
+  }
+
+  public ImmutableMap<String, ReplicationGroupOffset> getOffsets() {
+    return offsets;
+  }
+}