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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
