HBASE-27216 Revisit the ReplicationSyncUp tool

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

@@ -200,7 +200,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 {
@@ -676,16 +677,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 {
@@ -712,3 +710,26 @@ message ModifyStoreFileTrackerStateData {
 message ModifyColumnFamilyStoreFileTrackerStateData {
   required bytes family = 1;
 }
+
+enum AssignReplicationQueuesState {
+  ASSIGN_REPLICATION_QUEUES_ADD_MISSING_QUEUES = 1;
+  ASSIGN_REPLICATION_QUEUES_CLAIM = 2;
+}
+
+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

@@ -98,6 +98,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;
+  }
+}

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

@@ -0,0 +1,129 @@
+/*
+ * 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 java.util.Objects;
+import java.util.Optional;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+public class ReplicationQueueId {
+
+  private final ServerName serverName;
+
+  private final String peerId;
+
+  private final Optional<ServerName> sourceServerName;
+
+  public ReplicationQueueId(ServerName serverName, String peerId) {
+    this.serverName = Objects.requireNonNull(serverName);
+    this.peerId = Objects.requireNonNull(peerId);
+    this.sourceServerName = Optional.empty();
+  }
+
+  public ReplicationQueueId(ServerName serverName, String peerId, ServerName sourceServerName) {
+    this.serverName = Objects.requireNonNull(serverName);
+    this.peerId = Objects.requireNonNull(peerId);
+    this.sourceServerName = Optional.of(sourceServerName);
+  }
+
+  public ServerName getServerName() {
+    return serverName;
+  }
+
+  public String getPeerId() {
+    return peerId;
+  }
+
+  public Optional<ServerName> getSourceServerName() {
+    return sourceServerName;
+  }
+
+  public ServerName getServerWALsBelongTo() {
+    return sourceServerName.orElse(serverName);
+  }
+
+  public boolean isRecovered() {
+    return sourceServerName.isPresent();
+  }
+
+  public ReplicationQueueId claim(ServerName targetServerName) {
+    ServerName newSourceServerName = sourceServerName.orElse(serverName);
+    return new ReplicationQueueId(targetServerName, peerId, newSourceServerName);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(peerId, serverName, sourceServerName);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) {
+      return true;
+    }
+    if (!(obj instanceof ReplicationQueueId)) {
+      return false;
+    }
+    ReplicationQueueId other = (ReplicationQueueId) obj;
+    return Objects.equals(peerId, other.peerId) && Objects.equals(serverName, other.serverName)
+      && Objects.equals(sourceServerName, other.sourceServerName);
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder sb = new StringBuilder().append(peerId).append('-').append(serverName);
+    sourceServerName.ifPresent(s -> sb.append('\t').append(s.toString()));
+    return sb.toString();
+  }
+
+  public static ReplicationQueueId parse(String str) {
+    int dashIndex = str.indexOf('-');
+    String peerId = str.substring(0, dashIndex);
+    int tabIndex = str.indexOf('\t', dashIndex + 1);
+    if (tabIndex < 0) {
+      String serverName = str.substring(dashIndex + 1);
+      return new ReplicationQueueId(ServerName.valueOf(serverName), peerId);
+    } else {
+      String serverName = str.substring(dashIndex + 1, tabIndex);
+      String sourceServerName = str.substring(tabIndex + 1);
+      return new ReplicationQueueId(ServerName.valueOf(serverName), peerId,
+        ServerName.valueOf(sourceServerName));
+    }
+  }
+
+  public static String getPeerId(String str) {
+    int dashIndex = str.indexOf('-');
+    return str.substring(0, dashIndex);
+  }
+
+  public static byte[] getScanPrefix(ServerName serverName, String peerId) {
+    return Bytes.toBytes(peerId + "-" + serverName.toString());
+  }
+
+  public static byte[] getScanPrefix(String peerId) {
+    return Bytes.toBytes(peerId + "-");
+  }
+
+  public static byte[] getScanStartRowForNextPeerId(String peerId) {
+    // '.' is the next char after '-'
+    return Bytes.toBytes(peerId + ".");
+  }
+}