Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-27463 Reset sizeOfLogQueue when refresh replication source (#4863) 

Co-authored-by: huiruan <huiruan@tencent.com>
Signed-off-by: Duo Zhang <zhangduo@apache.org>
Reviewed-by: Rushabh Shah <shahrs87@gmail.com>
This commit is contained in:
Ruanhui 2022-11-27 23:07:40 +08:00 committed by GitHub
parent a5ff289d7a
commit bb9f43c6f9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 5 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
View File

@ -461,15 +461,16 @@ public class ReplicationSourceManager {
String terminateMessage = "Peer " + peerId
+ " state or config changed. Will close the previous replication source and open a new one";
ReplicationPeer peer = replicationPeers.getPeer(peerId);
ReplicationSourceInterface src = createSource(peerId, peer);
ReplicationSourceInterface src;
// synchronized on latestPaths to avoid missing the new log
synchronized (this.latestPaths) {
ReplicationSourceInterface toRemove = this.sources.put(peerId, src);
ReplicationSourceInterface toRemove = this.sources.remove(peerId);
if (toRemove != null) {
LOG.info("Terminate replication source for " + toRemove.getPeerId());
// Do not clear metrics
toRemove.terminate(terminateMessage, null, false);
toRemove.terminate(terminateMessage, null, true);
}
src = createSource(peerId, peer);
this.sources.put(peerId, src);
for (NavigableSet<String> walsByGroup : walsById.get(peerId).values()) {
walsByGroup.forEach(wal -> src.enqueueLog(new Path(this.logDir, wal)));
}