HBASE-27463 Reset sizeOfLogQueue when refresh replication source (apa…

…che#4863) Co-authored-by: huiruan <[email protected]> Signed-off-by: Duo Zhang <[email protected]> Reviewed-by: Rushabh Shah <[email protected]>
vinayakphegde · Nov 27, 2022 · bb9f43c · bb9f43c
1 parent a5ff289
commit bb9f43c
Showing 1 changed file with 5 additions and 4 deletions.
diff --git a/.../main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java b/.../main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
@@ -461,15 +461,16 @@ public void refreshSources(String peerId) throws IOException {
     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)));
       }