From 3152d9981b8d99ba8a7b72672275baab5faa52e7 Mon Sep 17 00:00:00 2001
From: binlijin <binlijin@gmail.com>
Date: Thu, 31 Oct 2019 09:23:59 +0800
Subject: [PATCH] HBASE-23231 ReplicationSource do not update metrics after
 refresh (#778)

Signed-off-by: stack <stack@apache.org>
Signed-off-by: Duo Zhang <zhangduo@apache.org>
---
 .../regionserver/ReplicationSource.java           | 15 ++++++++++++---
 .../regionserver/ReplicationSourceInterface.java  |  8 ++++++++
 .../regionserver/ReplicationSourceManager.java    |  3 ++-
 .../hbase/replication/ReplicationSourceDummy.java |  9 ++++++++-
 4 files changed, 30 insertions(+), 5 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
index 948c24dc4f86..78edffaf471a 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
@@ -549,7 +549,12 @@ public void terminate(String reason, Exception cause) {
     terminate(reason, cause, true);
   }
 
-  public void terminate(String reason, Exception cause, boolean join) {
+  @Override
+  public void terminate(String reason, Exception cause, boolean clearMetrics) {
+    terminate(reason, cause, clearMetrics, true);
+  }
+
+  public void terminate(String reason, Exception cause, boolean clearMetrics, boolean join) {
     if (cause == null) {
       LOG.info("{} Closing source {} because: {}", logPeerId(), this.queueId, reason);
     } else {
@@ -595,7 +600,9 @@ public void terminate(String reason, Exception cause, boolean join) {
     if (this.replicationEndpoint != null) {
       this.replicationEndpoint.stop();
     }
-    metrics.clear();
+    if (clearMetrics) {
+      metrics.clear();
+    }
     if (join) {
       for (ReplicationSourceShipper worker : workers) {
         Threads.shutdown(worker, this.sleepForRetries);
@@ -611,7 +618,9 @@ public void terminate(String reason, Exception cause, boolean join) {
         }
       }
     }
-    this.metrics.clear();
+    if (clearMetrics) {
+      this.metrics.clear();
+    }
   }
 
   @Override
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
index df7a8cc7b2ff..0bd90cf1ee89 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
@@ -91,6 +91,14 @@ void addHFileRefs(TableName tableName, byte[] family, List<Pair<Path, Path>> pai
    */
   void terminate(String reason, Exception cause);
 
+  /**
+   * End the replication
+   * @param reason why it's terminating
+   * @param cause the error that's causing it
+   * @param clearMetrics removes all metrics about this Source
+   */
+  void terminate(String reason, Exception cause, boolean clearMetrics);
+
   /**
    * Get the current log that's replicated
    * @return the current log
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
index de37cc862e7e..43afa798f08b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
@@ -482,7 +482,8 @@ public void refreshSources(String peerId) throws IOException {
       ReplicationSourceInterface toRemove = this.sources.put(peerId, src);
       if (toRemove != null) {
         LOG.info("Terminate replication source for " + toRemove.getPeerId());
-        toRemove.terminate(terminateMessage);
+        // Do not clear metrics
+        toRemove.terminate(terminateMessage, null, false);
       }
       for (NavigableSet<String> walsByGroup : walsById.get(peerId).values()) {
         walsByGroup.forEach(wal -> src.enqueueLog(new Path(this.logDir, wal)));
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java
index 67f793d628c5..a361c4470604 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java
@@ -86,7 +86,14 @@ public void terminate(String reason) {
 
   @Override
   public void terminate(String reason, Exception e) {
-    this.metrics.clear();
+    terminate(reason, e, true);
+  }
+
+  @Override
+  public void terminate(String reason, Exception e, boolean clearMetrics) {
+    if (clearMetrics) {
+      this.metrics.clear();
+    }
   }
 
   @Override