Apache9 commented on code in PR #4810:
URL: https://github.com/apache/hbase/pull/4810#discussion_r1016692615
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/DumpReplicationQueues.java:
##########
@@ -294,80 +348,114 @@ public String
dumpPeersState(List<ReplicationPeerDescription> peers) throws Exce
return sb.toString();
}
- public String dumpQueues(ZKWatcher zkw, Set<String> peerIds, boolean hdfs)
throws Exception {
- ReplicationQueueStorage queueStorage;
+ public String dumpQueues(Connection connection, Set<String> peerIds, boolean
hdfs,
+ Configuration conf) throws Exception {
StringBuilder sb = new StringBuilder();
+ ReplicationQueueStorage queueStorage =
+ ReplicationStorageFactory.getReplicationQueueStorage(connection, conf);
+
+ Set<ServerName> liveRegionServers =
+
connection.getAdmin().getClusterMetrics().getLiveServerMetrics().keySet();
- // queueStorage =
ReplicationStorageFactory.getReplicationQueueStorage(zkw, getConf());
- // Set<ServerName> liveRegionServers = ZKUtil.listChildrenNoWatch(zkw,
- // zkw.getZNodePaths().rsZNode)
- // .stream().map(ServerName::parseServerName).collect(Collectors.toSet());
- //
- // Loops each peer on each RS and dumps the queues
- // List<ServerName> regionservers = queueStorage.getListOfReplicators();
- // if (regionservers == null || regionservers.isEmpty()) {
- // return sb.toString();
- // }
- // for (ServerName regionserver : regionservers) {
- // List<String> queueIds = queueStorage.getAllQueues(regionserver);
- // if (!liveRegionServers.contains(regionserver)) {
- // deadRegionServers.add(regionserver.getServerName());
- // }
- // for (String queueId : queueIds) {
- // ReplicationQueueInfo queueInfo = new ReplicationQueueInfo(queueId);
- // List<String> wals = queueStorage.getWALsInQueue(regionserver, queueId);
- // Collections.sort(wals);
- // if (!peerIds.contains(queueInfo.getPeerId())) {
- // deletedQueues.add(regionserver + "/" + queueId);
- // sb.append(formatQueue(regionserver, queueStorage, queueInfo, queueId,
wals, true, hdfs));
- // } else {
- // sb.append(formatQueue(regionserver, queueStorage, queueInfo, queueId,
wals, false, hdfs));
- // }
- // }
- // }
+ List<ServerName> regionServers = queueStorage.listAllReplicators();
+ if (regionServers == null || regionServers.isEmpty()) {
+ return sb.toString();
+ }
+ for (ServerName regionServer : regionServers) {
+ List<ReplicationQueueId> queueIds =
queueStorage.listAllQueueIds(regionServer);
+
+ if (!liveRegionServers.contains(regionServer)) {
+ deadRegionServers.add(regionServer.getServerName());
+ }
+ for (ReplicationQueueId queueId : queueIds) {
+ List<String> tmpWals = new ArrayList<>();
+ // wals
+ tmpWals.addAll(AbstractFSWALProvider
+ .getWALFiles(connection.getConfiguration(),
+ URLEncoder.encode(queueId.getServerWALsBelongTo().toString(),
+ StandardCharsets.UTF_8.name()))
+ .stream().map(Path::toString).collect(Collectors.toList()));
+
+ // old wals
+ tmpWals.addAll(AbstractFSWALProvider
+ .getArchivedWALFiles(connection.getConfiguration(),
queueId.getServerWALsBelongTo(),
+ URLEncoder.encode(queueId.getServerWALsBelongTo().toString(),
+ StandardCharsets.UTF_8.name()))
+ .stream().map(Path::toString).collect(Collectors.toList()));
Review Comment:
Ditto.
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/DumpReplicationQueues.java:
##########
@@ -294,80 +348,114 @@ public String
dumpPeersState(List<ReplicationPeerDescription> peers) throws Exce
return sb.toString();
}
- public String dumpQueues(ZKWatcher zkw, Set<String> peerIds, boolean hdfs)
throws Exception {
- ReplicationQueueStorage queueStorage;
+ public String dumpQueues(Connection connection, Set<String> peerIds, boolean
hdfs,
+ Configuration conf) throws Exception {
StringBuilder sb = new StringBuilder();
+ ReplicationQueueStorage queueStorage =
+ ReplicationStorageFactory.getReplicationQueueStorage(connection, conf);
+
+ Set<ServerName> liveRegionServers =
+
connection.getAdmin().getClusterMetrics().getLiveServerMetrics().keySet();
- // queueStorage =
ReplicationStorageFactory.getReplicationQueueStorage(zkw, getConf());
- // Set<ServerName> liveRegionServers = ZKUtil.listChildrenNoWatch(zkw,
- // zkw.getZNodePaths().rsZNode)
- // .stream().map(ServerName::parseServerName).collect(Collectors.toSet());
- //
- // Loops each peer on each RS and dumps the queues
- // List<ServerName> regionservers = queueStorage.getListOfReplicators();
- // if (regionservers == null || regionservers.isEmpty()) {
- // return sb.toString();
- // }
- // for (ServerName regionserver : regionservers) {
- // List<String> queueIds = queueStorage.getAllQueues(regionserver);
- // if (!liveRegionServers.contains(regionserver)) {
- // deadRegionServers.add(regionserver.getServerName());
- // }
- // for (String queueId : queueIds) {
- // ReplicationQueueInfo queueInfo = new ReplicationQueueInfo(queueId);
- // List<String> wals = queueStorage.getWALsInQueue(regionserver, queueId);
- // Collections.sort(wals);
- // if (!peerIds.contains(queueInfo.getPeerId())) {
- // deletedQueues.add(regionserver + "/" + queueId);
- // sb.append(formatQueue(regionserver, queueStorage, queueInfo, queueId,
wals, true, hdfs));
- // } else {
- // sb.append(formatQueue(regionserver, queueStorage, queueInfo, queueId,
wals, false, hdfs));
- // }
- // }
- // }
+ List<ServerName> regionServers = queueStorage.listAllReplicators();
+ if (regionServers == null || regionServers.isEmpty()) {
+ return sb.toString();
+ }
+ for (ServerName regionServer : regionServers) {
+ List<ReplicationQueueId> queueIds =
queueStorage.listAllQueueIds(regionServer);
+
+ if (!liveRegionServers.contains(regionServer)) {
+ deadRegionServers.add(regionServer.getServerName());
+ }
+ for (ReplicationQueueId queueId : queueIds) {
+ List<String> tmpWals = new ArrayList<>();
+ // wals
+ tmpWals.addAll(AbstractFSWALProvider
+ .getWALFiles(connection.getConfiguration(),
+ URLEncoder.encode(queueId.getServerWALsBelongTo().toString(),
+ StandardCharsets.UTF_8.name()))
+ .stream().map(Path::toString).collect(Collectors.toList()));
Review Comment:
Just use forEach(tmpWals::add) to finish the stream?
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java:
##########
@@ -381,6 +383,28 @@ public static List<Path> getArchivedWALFiles(Configuration
conf, ServerName serv
return archivedWalFiles;
}
+ /**
+ * List all the wal files for a logPrefix.
+ */
+ public static List<Path> getWALFiles(Configuration c, String logPrefix)
throws IOException {
+ Path walRoot = new Path(CommonFSUtils.getWALRootDir(c),
HConstants.HREGION_LOGDIR_NAME);
+ FileSystem fs = walRoot.getFileSystem(c);
+ List<Path> walFiles = new ArrayList<>();
+ try {
+ RemoteIterator<LocatedFileStatus> it = fs.listFiles(walRoot, true);
Review Comment:
I do not think we need recursive list here? All wals for a live regionserver
will be placed under a special directory, just list the directory is enough...
--
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]
