2005hithlj commented on code in PR #4966:
URL: https://github.com/apache/hbase/pull/4966#discussion_r1100073675
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSyncUp.java:
##########
@@ -69,41 +80,103 @@ public static void main(String[] args) throws Exception {
System.exit(ret);
}
- private Set<ServerName> getLiveRegionServers(ZKWatcher zkw) throws
KeeperException {
- List<String> rsZNodes = ZKUtil.listChildrenNoWatch(zkw,
zkw.getZNodePaths().rsZNode);
- return rsZNodes == null
- ? Collections.emptySet()
- :
rsZNodes.stream().map(ServerName::parseServerName).collect(Collectors.toSet());
+ // Find region servers under wal directory
+ // Here we only care about the region servers which may still be alive, as
we need to add
+ // replications for them if missing. The dead region servers which have
already been processed
+ // fully do not need to add their replication queues again, as the operation
has already been done
+ // in SCP.
+ private Set<ServerName> listRegionServers(FileSystem walFs, Path walDir)
throws IOException {
+ FileStatus[] statuses;
+ try {
+ statuses = walFs.listStatus(walDir);
+ } catch (FileNotFoundException e) {
+ System.out.println("WAL directory " + walDir + " does not exists,
ignore");
+ return Collections.emptySet();
+ }
+ Set<ServerName> regionServers = new HashSet<>();
+ for (FileStatus status : statuses) {
+ // All wal files under the walDir is within its region server's directory
+ if (!status.isDirectory()) {
+ continue;
+ }
+ ServerName sn =
AbstractFSWALProvider.getServerNameFromWALDirectoryName(status.getPath());
+ if (sn != null) {
+ regionServers.add(sn);
+ }
+ }
+ return regionServers;
+ }
+
+ private void addMissingReplicationQueues(ReplicationQueueStorage storage,
ServerName regionServer,
+ Set<String> peerIds) throws ReplicationException {
+ Set<String> existingQueuePeerIds = new HashSet<>();
+ List<ReplicationQueueId> queueIds = storage.listAllQueueIds(regionServer);
+ for (Iterator<ReplicationQueueId> iter = queueIds.iterator();
iter.hasNext();) {
+ ReplicationQueueId queueId = iter.next();
+ if (!queueId.isRecovered()) {
+ existingQueuePeerIds.add(queueId.getPeerId());
+ }
+ }
+
+ for (String peerId : peerIds) {
+ if (!existingQueuePeerIds.contains(peerId)) {
+ ReplicationQueueId queueId = new ReplicationQueueId(regionServer,
peerId);
+ System.out.println("Add replication queue " + queueId + " for
claiming");
+ storage.setOffset(queueId, regionServer.toString(),
ReplicationGroupOffset.BEGIN,
+ Collections.emptyMap());
+ }
+ }
+ }
+
+ private void addMissingReplicationQueues(ReplicationQueueStorage storage,
+ Set<ServerName> regionServers, Set<String> peerIds) throws
ReplicationException {
+ for (ServerName regionServer : regionServers) {
+ addMissingReplicationQueues(storage, regionServer, peerIds);
+ }
}
// When using this tool, usually the source cluster is unhealthy, so we
should try to claim the
// replication queues for the dead region servers first and then replicate
the data out.
- private void claimReplicationQueues(ZKWatcher zkw, ReplicationSourceManager
mgr)
- throws ReplicationException, KeeperException {
- // TODO: reimplement this tool
- // List<ServerName> replicators =
mgr.getQueueStorage().getListOfReplicators();
- // Set<ServerName> liveRegionServers = getLiveRegionServers(zkw);
- // for (ServerName sn : replicators) {
- // if (!liveRegionServers.contains(sn)) {
- // List<String> replicationQueues = mgr.getQueueStorage().getAllQueues(sn);
- // System.out.println(sn + " is dead, claim its replication queues: " +
replicationQueues);
- // for (String queue : replicationQueues) {
- // mgr.claimQueue(sn, queue);
- // }
- // }
- // }
+ private void claimReplicationQueues(ReplicationSourceManager mgr,
Set<ServerName> regionServers)
+ throws ReplicationException, KeeperException, IOException {
+ // union the region servers from both places, i.e, from the wal directory,
and the records in
+ // replication queue storage.
+ Set<ServerName> replicators = new HashSet<>(regionServers);
+ ReplicationQueueStorage queueStorage = mgr.getQueueStorage();
+ replicators.addAll(queueStorage.listAllReplicators());
+ FileSystem fs = CommonFSUtils.getCurrentFileSystem(getConf());
+ Path infoDir = new Path(CommonFSUtils.getRootDir(getConf()), INFO_DIR);
+ for (ServerName sn : replicators) {
+ List<ReplicationQueueId> replicationQueues =
queueStorage.listAllQueueIds(sn);
+ System.out.println(sn + " is dead, claim its replication queues: " +
replicationQueues);
+ // record the rs name, so when master restarting, we will skip claiming
its replication queue
+ fs.createNewFile(new Path(infoDir, sn.getServerName()));
+ for (ReplicationQueueId queueId : replicationQueues) {
+ mgr.claimQueue(queueId, true);
+ }
+ }
}
@Override
public int run(String[] args) throws Exception {
Review Comment:
Whether there is a problem with starting two ReplicationSyncUp processes at
the same time?
There is no check in the code at present.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
