wchevreuil commented on a change in pull request #2649:
URL: https://github.com/apache/hbase/pull/2649#discussion_r522201941
##########
File path:
hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationQueueInfo.java
##########
@@ -57,14 +63,30 @@ 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
* cases: 2-ip-10-46-221-101.ec2.internal,52170,1364333181125-<server
name>-...
*/
- private static void
- extractDeadServersFromZNodeString(String deadServerListStr,
List<ServerName> result) {
- if(deadServerListStr == null || result == null ||
deadServerListStr.isEmpty()) {
+ private void extractDeadServersFromZNodeString(String deadServerListStr,
Review comment:
no need to be static anymore?
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/master/ReplicationServerManager.java
##########
@@ -179,26 +177,81 @@ public void expireServer(final ServerName serverName) {
* @return ServerMetrics if serverName is known else null
*/
public ServerMetrics getServerMetrics(final ServerName serverName) {
- return this.onlineServers.get(serverName);
+ if (!this.onlineServers.containsKey(serverName)) {
+ return null;
+ }
+ return this.onlineServers.get(serverName).getFirst();
}
- private class OnlineServerRefresher extends ScheduledChore {
+ /**
+ * This chore is responsible for 3 things:
+ * 1. Find all alive replication servers.
+ * 2. Find all replication queues.
+ * 3. Assign different queue to different replication server.
+ */
+ private class ReplicationServerRefresher extends ScheduledChore {
- public OnlineServerRefresher(String name, int p) {
- super(name, master, p, 60 * 1000); // delay one minute before first
execute
+ public ReplicationServerRefresher(String name, int p) {
+ super(name, master, p, p);
}
@Override
protected void chore() {
+ // Find all alive replication servers
synchronized (onlineServers) {
List<ServerName> servers = getOnlineServersList();
servers.forEach(s -> {
- ServerMetrics metrics = onlineServers.get(s);
+ ServerMetrics metrics = onlineServers.get(s).getFirst();
if (metrics.getReportTimestamp() + refreshPeriod <
System.currentTimeMillis()) {
expireServer(s);
}
});
}
+ Set<String> assignedQueueNodes =
+
onlineServers.values().stream().map(Pair::getSecond).flatMap(Set::stream)
+ .collect(Collectors.toSet());
+ Map<ServerName, Set<String>> unassigned = new HashMap<>();
+ // Because all replication queues is owned by region servers. List all
region servers and get
+ // their replication queues.
+ for (ServerName producer :
master.getServerManager().getOnlineServersList()) {
+ try {
+ List<String> queues = zkQueueStorage.getAllQueues(producer);
+ for (String queue : queues) {
+ String queueNode = zkQueueStorage.getQueueNode(producer, queue);
+ LOG.debug("Found one replication queue {}", queueNode);
+ if (!assignedQueueNodes.contains(queueNode)) {
+ unassigned.computeIfAbsent(producer, p -> new
HashSet<>()).add(queue);
+ }
+ }
+ } catch (ReplicationException e) {
+ LOG.warn("Failed to get all replication queues of server {}",
producer, e);
+ }
+ }
+ ServerName[] consumers =
getOnlineServersList().stream().toArray(ServerName[]::new);
+ if (consumers.length == 0) {
+ LOG.warn("No replication server available!");
+ return;
+ }
+ // Assign different queue to different replication server
+ for (Map.Entry<ServerName, Set<String>> entry : unassigned.entrySet()) {
+ ServerName producer = entry.getKey();
+ for (String queueId : entry.getValue()) {
+ ServerName consumer =
consumers[ThreadLocalRandom.current().nextInt(consumers.length)];
+ ReplicationServerProtos.StartReplicationSourceRequest request =
+ ReplicationServerProtos.StartReplicationSourceRequest.newBuilder()
+
.setServerName(ProtobufUtil.toServerName(producer)).setQueueId(queueId).build();
+ try {
+
FutureUtils.get(master.getAsyncClusterConnection().getReplicationServerAdmin(consumer)
+ .startReplicationSource(request, 10000));
Review comment:
Is `startReplicationSource` going to update zk queue for the original
server that we found to be unassigned?
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HReplicationServer.java
##########
@@ -266,13 +270,19 @@ public void run() {
online.set(true);
- long lastMsg = System.currentTimeMillis();
+ int refreshPeriod = conf.getInt(REPLICATION_SERVER_REFRESH_PERIOD,
+ REPLICATION_SERVER_REFRESH_PERIOD_DEFAULT);
+ long lastReportedTime = System.currentTimeMillis();
// The main run loop.
while (!isStopped()) {
long now = System.currentTimeMillis();
- if ((now - lastMsg) >= msgInterval) {
- tryReplicationServerReport(lastMsg, now);
- lastMsg = System.currentTimeMillis();
+ if ((now - lastReportedTime) >= msgInterval) {
+ if (tryReplicationServerReport(lastReportedTime, now)) {
Review comment:
In the case of a ReplicationServer failure and another ReplicationServer
picking up the queue, does ReplicationManager get notified right away, or will
it have to wait until the time of next report?
##########
File path:
hbase-protocol-shaded/src/main/protobuf/server/master/ReplicationServerStatus.proto
##########
@@ -25,10 +25,24 @@ 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";
+
+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;
Review comment:
Should be simply "queues"? Got a bit confused by the term "queueNodes"
further down this code path.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
