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

Co-authored-by: huiruan <[email protected]>
Signed-off-by: Duo Zhang <[email protected]>
Reviewed-by: Rushabh Shah <[email protected]>
(cherry picked from commit bb9f43c)

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

@@ -400,22 +400,18 @@ 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);
       }
-      for (SortedSet<String> walsByGroup : walsById.get(peerId).values()) {
-        walsByGroup.forEach(wal -> {
-          Path walPath = new Path(this.logDir, wal);
-          src.enqueueLog(walPath);
-          LOG.trace("Enqueued {} to source {} during source creation.", walPath, src.getQueueId());
-        });
-
+      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)));
       }
     }
     LOG.info("Startup replication source for " + src.getPeerId());