You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hbase.apache.org by GitBox <gi...@apache.org> on 2022/10/19 16:02:10 UTC
[GitHub] [hbase] Apache9 commented on a diff in pull request #4810: HBASE-27217 Revisit the DumpReplicationQueues tool
Apache9 commented on code in PR #4810:
URL: https://github.com/apache/hbase/pull/4810#discussion_r999649192
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/DumpReplicationQueues.java:
##########
@@ -229,21 +238,45 @@ private int dumpReplicationQueues(DumpOptions opts) throws Exception {
LOG.info("Found [--distributed], will poll each RegionServer.");
Set<String> peerIds =
peers.stream().map((peer) -> peer.getPeerId()).collect(Collectors.toSet());
- System.out.println(dumpQueues(zkw, peerIds, opts.isHdfs()));
+ System.out.println(dumpQueues(zkw, connection, peerIds, opts.isHdfs()));
Review Comment:
We still need zkw?
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/DumpReplicationQueues.java:
##########
@@ -294,80 +327,107 @@ 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(ZKWatcher zkw, Connection connection, Set<String> peerIds, boolean hdfs)
+ throws Exception {
StringBuilder sb = new StringBuilder();
+ ReplicationQueueStorage queueStorage =
+ ReplicationStorageFactory.getReplicationQueueStorage(connection, getConf());
+ Set<ServerName> liveRegionServers = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().rsZNode)
+ .stream().map(ServerName::parseServerName).collect(Collectors.toSet());
+
+ List<ServerName> regionServers = queueStorage.listAllReplicators();
+ if (regionServers == null || regionServers.isEmpty()) {
+ return sb.toString();
+ }
+ for (ServerName regionServer : regionServers) {
+ List<ReplicationQueueId> queueIds = queueStorage.listAllQueueIds(regionServer);
- // 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));
- // }
- // }
- // }
+ if (!liveRegionServers.contains(regionServer)) {
+ deadRegionServers.add(regionServer.getServerName());
+ }
+ for (ReplicationQueueId queueId : queueIds) {
+ List<String> wals = null;
+ if (queueId.isRecovered()) {
+ wals = AbstractFSWALProvider
Review Comment:
Just use queueId.getServerWALsBelongTo?
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/DumpReplicationQueues.java:
##########
@@ -294,80 +327,107 @@ 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(ZKWatcher zkw, Connection connection, Set<String> peerIds, boolean hdfs)
+ throws Exception {
StringBuilder sb = new StringBuilder();
+ ReplicationQueueStorage queueStorage =
+ ReplicationStorageFactory.getReplicationQueueStorage(connection, getConf());
+ Set<ServerName> liveRegionServers = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().rsZNode)
Review Comment:
This could also be done through Admin interface, so we do not need to depend on zk?
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/DumpReplicationQueues.java:
##########
@@ -294,80 +327,107 @@ 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(ZKWatcher zkw, Connection connection, Set<String> peerIds, boolean hdfs)
+ throws Exception {
StringBuilder sb = new StringBuilder();
+ ReplicationQueueStorage queueStorage =
+ ReplicationStorageFactory.getReplicationQueueStorage(connection, getConf());
+ Set<ServerName> liveRegionServers = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().rsZNode)
+ .stream().map(ServerName::parseServerName).collect(Collectors.toSet());
+
+ List<ServerName> regionServers = queueStorage.listAllReplicators();
+ if (regionServers == null || regionServers.isEmpty()) {
+ return sb.toString();
+ }
+ for (ServerName regionServer : regionServers) {
+ List<ReplicationQueueId> queueIds = queueStorage.listAllQueueIds(regionServer);
- // 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));
- // }
- // }
- // }
+ if (!liveRegionServers.contains(regionServer)) {
+ deadRegionServers.add(regionServer.getServerName());
+ }
+ for (ReplicationQueueId queueId : queueIds) {
+ List<String> wals = null;
+ if (queueId.isRecovered()) {
+ wals = AbstractFSWALProvider
+ .getArchivedWALFiles(connection.getConfiguration(), queueId.getSourceServerName().get(),
+ queueId.getSourceServerName().get().toString())
+ .stream().map(Path::toString).collect(Collectors.toList());
+ } else {
+ wals = AbstractFSWALProvider
Review Comment:
Is this enough to get all the wals for a live region server?
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/DumpReplicationQueues.java:
##########
@@ -294,80 +327,107 @@ 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(ZKWatcher zkw, Connection connection, Set<String> peerIds, boolean hdfs)
+ throws Exception {
StringBuilder sb = new StringBuilder();
+ ReplicationQueueStorage queueStorage =
+ ReplicationStorageFactory.getReplicationQueueStorage(connection, getConf());
+ Set<ServerName> liveRegionServers = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().rsZNode)
+ .stream().map(ServerName::parseServerName).collect(Collectors.toSet());
+
+ List<ServerName> regionServers = queueStorage.listAllReplicators();
+ if (regionServers == null || regionServers.isEmpty()) {
+ return sb.toString();
+ }
+ for (ServerName regionServer : regionServers) {
+ List<ReplicationQueueId> queueIds = queueStorage.listAllQueueIds(regionServer);
- // 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));
- // }
- // }
- // }
+ if (!liveRegionServers.contains(regionServer)) {
+ deadRegionServers.add(regionServer.getServerName());
+ }
+ for (ReplicationQueueId queueId : queueIds) {
+ List<String> wals = null;
+ if (queueId.isRecovered()) {
+ wals = AbstractFSWALProvider
+ .getArchivedWALFiles(connection.getConfiguration(), queueId.getSourceServerName().get(),
+ queueId.getSourceServerName().get().toString())
+ .stream().map(Path::toString).collect(Collectors.toList());
+ } else {
+ wals = AbstractFSWALProvider
+ .getArchivedWALFiles(connection.getConfiguration(), queueId.getServerName(),
+ queueId.getServerName().toString())
+ .stream().map(Path::toString).collect(Collectors.toList());
+ }
+ Collections.sort(wals);
Review Comment:
String compare is enough for sorting? Not sure...
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/DumpReplicationQueues.java:
##########
@@ -294,80 +327,107 @@ 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(ZKWatcher zkw, Connection connection, Set<String> peerIds, boolean hdfs)
+ throws Exception {
StringBuilder sb = new StringBuilder();
+ ReplicationQueueStorage queueStorage =
+ ReplicationStorageFactory.getReplicationQueueStorage(connection, getConf());
+ Set<ServerName> liveRegionServers = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().rsZNode)
+ .stream().map(ServerName::parseServerName).collect(Collectors.toSet());
+
+ List<ServerName> regionServers = queueStorage.listAllReplicators();
+ if (regionServers == null || regionServers.isEmpty()) {
+ return sb.toString();
+ }
+ for (ServerName regionServer : regionServers) {
+ List<ReplicationQueueId> queueIds = queueStorage.listAllQueueIds(regionServer);
- // 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));
- // }
- // }
- // }
+ if (!liveRegionServers.contains(regionServer)) {
+ deadRegionServers.add(regionServer.getServerName());
+ }
+ for (ReplicationQueueId queueId : queueIds) {
+ List<String> wals = null;
+ if (queueId.isRecovered()) {
+ wals = AbstractFSWALProvider
+ .getArchivedWALFiles(connection.getConfiguration(), queueId.getSourceServerName().get(),
+ queueId.getSourceServerName().get().toString())
+ .stream().map(Path::toString).collect(Collectors.toList());
+ } else {
+ wals = AbstractFSWALProvider
+ .getArchivedWALFiles(connection.getConfiguration(), queueId.getServerName(),
+ queueId.getServerName().toString())
+ .stream().map(Path::toString).collect(Collectors.toList());
+ }
+ Collections.sort(wals);
+ if (!peerIds.contains(queueId.getPeerId())) {
+ deletedQueues.add(regionServer + "/" + queueId);
+ sb.append(formatQueue(regionServer, queueStorage, wals, queueId, true, hdfs));
+ } else {
+ sb.append(formatQueue(regionServer, queueStorage, wals, queueId, false, hdfs));
+ }
+ }
+ }
return sb.toString();
}
- private String formatQueue(ServerName regionserver, ReplicationQueueStorage queueStorage,
- ReplicationQueueInfo queueInfo, String queueId, List<String> wals, boolean isDeleted,
- boolean hdfs) throws Exception {
+ private String formatQueue(ServerName regionServer, ReplicationQueueStorage queueStorage,
+ List<String> wals, ReplicationQueueId queueId, boolean isDeleted, boolean hdfs)
+ throws Exception {
StringBuilder sb = new StringBuilder();
- List<ServerName> deadServers;
-
- sb.append("Dumping replication queue info for RegionServer: [" + regionserver + "]" + "\n");
+ sb.append("Dumping replication queue info for RegionServer: [" + regionServer + "]" + "\n");
sb.append(" Queue znode: " + queueId + "\n");
- sb.append(" PeerID: " + queueInfo.getPeerId() + "\n");
- sb.append(" Recovered: " + queueInfo.isQueueRecovered() + "\n");
- deadServers = queueInfo.getDeadRegionServers();
- if (deadServers.isEmpty()) {
- sb.append(" No dead RegionServers found in this queue." + "\n");
+ sb.append(" PeerID: " + queueId.getPeerId() + "\n");
+ sb.append(" Recovered: " + queueId.isRecovered() + "\n");
+ // In new version, we only record the first dead RegionServer in queueId.
+ if (queueId.getSourceServerName().isPresent()) {
+ sb.append(" Dead RegionServer: " + queueId.getSourceServerName().get() + "\n");
} else {
- sb.append(" Dead RegionServers: " + deadServers + "\n");
+ sb.append(" No dead RegionServer found in this queue." + "\n");
}
sb.append(" Was deleted: " + isDeleted + "\n");
sb.append(" Number of WALs in replication queue: " + wals.size() + "\n");
- peersQueueSize.addAndGet(queueInfo.getPeerId(), wals.size());
-
- for (String wal : wals) {
- // long position = queueStorage.getWALPosition(regionserver, queueInfo.getPeerId(), wal);
- // sb.append(" Replication position for " + wal + ": "
- // + (position > 0 ? position : "0" + " (not started or nothing to replicate)") + "\n");
+ peersQueueSize.addAndGet(queueId.getPeerId(), wals.size());
+
+ Set<Map.Entry<String, ReplicationGroupOffset>> offsets =
+ queueStorage.getOffsets(queueId).entrySet();
+
+ for (Map.Entry<String, ReplicationGroupOffset> entry : offsets) {
+ String walGroup = null;
+ walGroup = entry.getKey();
+ for (String wal : wals) {
+ ReplicationGroupOffset offset = entry.getValue();
+ if (offset.getWal().equals(wal)) {
+ long position = offset.getOffset();
+ sb.append(
+ " Replication position for " + (walGroup != null ? walGroup + "/" + wal : wal) + ": ");
+
+ // Position is -1, which means that the file has already been fully replicated,
+ // the logic here is different from the previous version.
+ if (position == -1) {
+ sb.append("-1 (has been replicated.)");
Review Comment:
I think we can just skip all the wals before the ReplicationGroupOffset?
--
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: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org