Apache9 commented on code in PR #4810:
URL: https://github.com/apache/hbase/pull/4810#discussion_r1003129402
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/DumpReplicationQueues.java:
##########
@@ -294,80 +321,117 @@ 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)
+ throws Exception {
StringBuilder sb = new StringBuilder();
+ ReplicationQueueStorage queueStorage =
+ ReplicationStorageFactory.getReplicationQueueStorage(connection,
getConf());
+
+ Set<ServerName> liveRegionServers =
+
connection.getAdmin().getClusterMetrics().getLiveServerMetrics().keySet();
+
+ 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) {
+ // wals
+ List<String> tmpWals = 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()));
+
+ Map<String, ReplicationGroupOffset> offsets =
queueStorage.getOffsets(queueId);
+ // filter out the wal files that should replicate
+ List<String> wals = new ArrayList<>();
+ for (Map.Entry<String, ReplicationGroupOffset> entry :
offsets.entrySet()) {
+ ReplicationGroupOffset offset = entry.getValue();
+ for (String wal : tmpWals) {
+ if (ReplicationOffsetUtil.shouldReplicate(offset, wal)) {
+ wals.add(wal);
+ }
+ }
+ }
+
+ Collections.sort(wals);
+ if (!peerIds.contains(queueId.getPeerId())) {
+ deletedQueues.add(regionServer + "/" + queueId);
+ sb.append(formatQueue(regionServer, offsets, wals, queueId, true,
hdfs));
+ } else {
+ sb.append(formatQueue(regionServer, offsets, 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, Map<String,
ReplicationGroupOffset> offsets,
+ 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(" 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("Dumping replication queue info for RegionServer: [" +
regionServer + "]" + "\n");
+ sb.append(" Queue id: " + queueId + "\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());
+
+ for (Map.Entry<String, ReplicationGroupOffset> entry : offsets.entrySet())
{
+ String walGroup = entry.getKey();
+ ReplicationGroupOffset offset = entry.getValue();
+ for (String wal : wals) {
+ 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) {
Review Comment:
Is it possible that we arrive here but position is -1? I think it will be
filtered out in the above ReplicationOffsetUtil.shouldReplicate check?
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/DumpReplicationQueues.java:
##########
@@ -229,21 +232,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(connection, peerIds, opts.isHdfs()));
System.out.println(dumpReplicationSummary());
} else {
- // use ZK instead
- System.out.print("Dumping replication znodes via ZooKeeper:");
- System.out.println(ZKDump.getReplicationZnodesDump(zkw));
+ // use replication table instead
+ System.out.println("Dumping replication info via replication table.");
+ System.out.println(dumpReplicationViaTable(connection));
}
return (0);
} catch (IOException e) {
return (-1);
} finally {
- zkw.close();
+ connection.close();
}
}
+ private String dumpReplicationViaTable(Connection connection) throws
ReplicationException {
Review Comment:
Checked the code in ZKDump.getReplicationZnodesDump, I do not think this is
a direct mapping of the code there?
In ZKDump.getReplicationZnodesDump, first we dump the replication peer data,
while here we do not dump it.
And when dumping the replication queue data, we will iterate all rsZNode and
print them all, but here we first print all the peers and region servers, and
then print all the wal files? Seems a bit strange...
And for dumping hfile refs, we will print hfile refs for each peer in
ZKDump.getReplicationZnodesDump, but here we just print them all, without the
peer information...
##########
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:
Better use TS_COMPARATOR?
##########
hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestDumpReplicationQueues.java:
##########
@@ -83,7 +83,7 @@ public void testDumpReplicationReturnsWalSorted() throws
Exception {
Set<String> peerIds = new HashSet<>();
peerIds.add("1");
dumpQueues.setConf(config);
- String dump = dumpQueues.dumpQueues(zkWatcherMock, peerIds, false);
+ String dump = dumpQueues.dumpQueues(null, peerIds, false);
Review Comment:
Parsing null is enough here? We do not need to pass a Connection instance?
--
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]
