HBASE-20524 Need to clear metrics when ReplicationSourceManager refresh replication sources

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

@@ -149,6 +149,7 @@ public class RecoveredReplicationSource extends ReplicationSource {
       Threads.sleep(100);// wait a short while for other worker thread to fully exit
       boolean allTasksDone = workerThreads.values().stream().allMatch(w -> w.isFinished());
       if (allTasksDone) {
+        this.getSourceMetrics().clear();
         manager.removeRecoveredSource(this);
         LOG.info("Finished recovering queue {} with the following stats: {}", queueId, getStats());
       }

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

@@ -519,6 +519,7 @@ public class ReplicationSource implements ReplicationSourceInterface {
         }
       }
     }
+    this.metrics.clear();
   }
 
   @Override

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

@@ -418,7 +418,6 @@ public class ReplicationSourceManager implements ReplicationListener {
    */
   void removeRecoveredSource(ReplicationSourceInterface src) {
     LOG.info("Done with the recovered queue " + src.getQueueId());
-    src.getSourceMetrics().clear();
     this.oldsources.remove(src);
     // Delete queue from storage and memory
     deleteQueue(src.getQueueId());
@@ -431,7 +430,6 @@ public class ReplicationSourceManager implements ReplicationListener {
    */
   void removeSource(ReplicationSourceInterface src) {
     LOG.info("Done with the queue " + src.getQueueId());
-    src.getSourceMetrics().clear();
     this.sources.remove(src.getPeerId());
     // Delete queue from storage and memory
     deleteQueue(src.getQueueId());

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java

@@ -80,12 +80,12 @@ public class ReplicationSourceDummy implements ReplicationSourceInterface {
 
   @Override
   public void terminate(String reason) {
-
+    terminate(reason, null);
   }
 
   @Override
   public void terminate(String reason, Exception e) {
-
+    this.metrics.clear();
   }
 
   @Override