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>
(cherry picked from commit bb9f43c6f9)
This commit is contained in:
Ruanhui 2022-11-27 23:07:40 +08:00 committed by Duo Zhang
parent 0a37292113
commit 5ce1d8f9bd
1 changed files with 7 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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