HBASE-25110 Add heartbeat for ReplicationServer and dispatch replication sources to ReplicationServer

Author: ddupg

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

@@ -25,10 +25,23 @@ option java_generic_services = true;
 option java_generate_equals_and_hash = true;
 option optimize_for = SPEED;
 
-import "server/master/RegionServerStatus.proto";
+import "HBase.proto";
+import "server/ClusterStatus.proto";
 
-service ReplicationServerStatusService {
+message ReplicationServerReportRequest {
+  required ServerName server = 1;
+
+  /** load the server is under */
+  optional ServerLoad load = 2;
+
+  /** The replication queues which this replication server is responsible for. */
+  repeated string queue_node = 3;
+}
 
-  rpc ReplicationServerReport(RegionServerReportRequest)
-      returns(RegionServerReportResponse);
+message ReplicationServerReportResponse {
+}
+
+service ReplicationServerStatusService {
+  rpc ReplicationServerReport(ReplicationServerReportRequest)
+    returns(ReplicationServerReportResponse);
 }

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

@@ -31,11 +31,16 @@
  * This class is responsible for the parsing logic for a queue id representing a queue.
  * It will extract the peerId if it's recovered as well as the dead region servers
  * that were part of the queue's history.
+ * One replication queue has only one owner. And the owner must be one region server. When enable
+ * replication offload feature, region server will not start replication source thread to replicate
+ * data. The replication queue will be used by replication server which is responsible for
+ * replicating data.
  */
 @InterfaceAudience.Private
 public class ReplicationQueueInfo {
   private static final Logger LOG = LoggerFactory.getLogger(ReplicationQueueInfo.class);
 
+  private final ServerName owner;
   private final String peerId;
   private final String queueId;
   private boolean queueRecovered;
@@ -46,7 +51,8 @@ public class ReplicationQueueInfo {
    * The passed queueId will be either the id of the peer or the handling story of that queue
    * in the form of id-servername-*
    */
-  public ReplicationQueueInfo(String queueId) {
+  public ReplicationQueueInfo(ServerName owner, String queueId) {
+    this.owner = owner;
     this.queueId = queueId;
     String[] parts = queueId.split("-", 2);
     this.queueRecovered = parts.length != 1;
@@ -57,6 +63,22 @@ public ReplicationQueueInfo(String queueId) {
     }
   }
 
+  /**
+   * A util method to parse the peerId from queueId.
+   */
+  public static String parsePeerId(String queueId) {
+    String[] parts = queueId.split("-", 2);
+    return parts.length != 1 ? parts[0] : queueId;
+  }
+
+  /**
+   * A util method to check whether a queue is recovered.
+   */
+  public static boolean isQueueRecovered(String queueId) {
+    String[] parts = queueId.split("-", 2);
+    return parts.length != 1;
+  }
+
   /**
    * Parse dead server names from queue id. servername can contain "-" such as
    * "ip-10-46-221-101.ec2.internal", so we need skip some "-" during parsing for the following
@@ -114,6 +136,10 @@ public List<ServerName> getDeadRegionServers() {
     return Collections.unmodifiableList(this.deadRegionServers);
   }
 
+  public ServerName getOwner() {
+    return this.owner;
+  }
+
   public String getPeerId() {
     return this.peerId;
   }

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

@@ -86,8 +86,7 @@ public static void removeAllQueues(ReplicationQueueStorage queueStorage, String
     for (ServerName replicator : queueStorage.getListOfReplicators()) {
       List<String> queueIds = queueStorage.getAllQueues(replicator);
       for (String queueId : queueIds) {
-        ReplicationQueueInfo queueInfo = new ReplicationQueueInfo(queueId);
-        if (queueInfo.getPeerId().equals(peerId)) {
+        if (ReplicationQueueInfo.parsePeerId(queueId).equals(peerId)) {
           queueStorage.removeQueue(replicator, queueId);
         }
       }

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

@@ -205,7 +205,7 @@ public void removeWAL(ServerName serverName, String queueId, String fileName)
 
   private void addLastSeqIdsToOps(String queueId, Map<String, Long> lastSeqIds,
       List<ZKUtilOp> listOfOps) throws KeeperException, ReplicationException {
-    String peerId = new ReplicationQueueInfo(queueId).getPeerId();
+    String peerId = ReplicationQueueInfo.parsePeerId(queueId);
     for (Entry<String, Long> lastSeqEntry : lastSeqIds.entrySet()) {
       String path = getSerialReplicationRegionPeerNode(lastSeqEntry.getKey(), peerId);
       Pair<Long, Integer> p = getLastSequenceIdWithVersion(lastSeqEntry.getKey(), peerId);

hbase-server/src/main/java/org/apache/hadoop/hbase/client/AsyncReplicationServerAdmin.java

@@ -29,6 +29,8 @@
 
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationServerProtos.ReplicationServerService;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationServerProtos.StartReplicationSourceRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationServerProtos.StartReplicationSourceResponse;
 
 /**
  * A simple wrapper of the {@link ReplicationServerService} for a replication server.
@@ -70,11 +72,23 @@ private <RESP> CompletableFuture<RESP> call(RpcCall<RESP> rpcCall, CellScanner c
     return future;
   }
 
+  private <RESP> CompletableFuture<RESP> call(RpcCall<RESP> rpcCall) {
+    return call(rpcCall, null);
+  }
+
   public CompletableFuture<AdminProtos.ReplicateWALEntryResponse> replicateWALEntry(
       AdminProtos.ReplicateWALEntryRequest request, CellScanner cellScanner, int timeout) {
     return call((stub, controller, done) -> {
       controller.setCallTimeout(timeout);
       stub.replicateWALEntry(controller, request, done);
     }, cellScanner);
   }
+
+  public CompletableFuture<StartReplicationSourceResponse> startReplicationSource(
+    StartReplicationSourceRequest request, int timeout) {
+    return call((stub, controller, done) -> {
+      controller.setCallTimeout(timeout);
+      stub.startReplicationSource(controller, request, done);
+    });
+  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java

@@ -75,7 +75,6 @@
 import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
 import org.apache.hadoop.hbase.ipc.ServerRpcController;
 import org.apache.hadoop.hbase.master.assignment.RegionStates;
-import org.apache.hadoop.hbase.master.assignment.TransitRegionStateProcedure;
 import org.apache.hadoop.hbase.master.janitor.MetaFixer;
 import org.apache.hadoop.hbase.master.locking.LockProcedure;
 import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
@@ -401,6 +400,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.TransitReplicationPeerSyncReplicationStateResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.UpdateReplicationPeerConfigRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.UpdateReplicationPeerConfigResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationServerStatusProtos.ReplicationServerReportRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationServerStatusProtos.ReplicationServerReportResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationServerStatusProtos.ReplicationServerStatusService;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos.SnapshotDescription;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.VisibilityLabelsProtos.VisibilityLabelsService;
@@ -3399,6 +3400,7 @@ private MasterProtos.BalancerDecisionsResponse getBalancerDecisions(
       .addAllBalancerDecision(balancerDecisions).build();
   }
 
+  @Override
   public ListReplicationSinkServersResponse listReplicationSinkServers(
     RpcController controller, ListReplicationSinkServersRequest request)
     throws ServiceException {
@@ -3408,7 +3410,7 @@ public ListReplicationSinkServersResponse listReplicationSinkServers(
       if (master.getMasterCoprocessorHost() != null) {
         master.getMasterCoprocessorHost().preListReplicationSinkServers();
       }
-      builder.addAllServerName(master.getReplicationServerManager().getOnlineServersList().stream()
+      builder.addAllServerName(master.listReplicationSinkServers().stream()
         .map(ProtobufUtil::toServerName).collect(Collectors.toList()));
       if (master.getMasterCoprocessorHost() != null) {
         master.getMasterCoprocessorHost().postListReplicationSinkServers();
@@ -3420,8 +3422,8 @@ public ListReplicationSinkServersResponse listReplicationSinkServers(
   }
 
   @Override
-  public RegionServerReportResponse replicationServerReport(RpcController controller,
-      RegionServerReportRequest request) throws ServiceException {
+  public ReplicationServerReportResponse replicationServerReport(RpcController controller,
+    ReplicationServerReportRequest request) throws ServiceException {
     try {
       master.checkServiceStarted();
       int versionNumber = 0;
@@ -3435,16 +3437,17 @@ public RegionServerReportResponse replicationServerReport(RpcController controll
       ServerName serverName = ProtobufUtil.toServerName(request.getServer());
       ServerMetrics oldMetrics = master.getReplicationServerManager().getServerMetrics(serverName);
       ServerMetrics newMetrics =
-          ServerMetricsBuilder.toServerMetrics(serverName, versionNumber, version, sl);
-      master.getReplicationServerManager().serverReport(serverName, newMetrics);
+        ServerMetricsBuilder.toServerMetrics(serverName, versionNumber, version, sl);
+      Set<String> queueNodes = request.getQueueNodeList().stream().collect(Collectors.toSet());
+      master.getReplicationServerManager().serverReport(serverName, newMetrics, queueNodes);
       if (sl != null && master.metricsMaster != null) {
         // Up our metrics.
         master.metricsMaster.incrementRequests(sl.getTotalNumberOfRequests()
-            - (oldMetrics != null ? oldMetrics.getRequestCount() : 0));
+          - (oldMetrics != null ? oldMetrics.getRequestCount() : 0));
       }
     } catch (IOException ioe) {
       throw new ServiceException(ioe);
     }
-    return RegionServerReportResponse.newBuilder().build();
+    return ReplicationServerReportResponse.newBuilder().build();
   }
 }