From 840c62e532021501c3a85241fae8cb81991e34df Mon Sep 17 00:00:00 2001
From: Yuta Imazu <airman515@gmail.com>
Date: Tue, 28 Feb 2023 14:13:49 +0900
Subject: [PATCH] HBASE-24781 Clean up peer metrics when disabling peer (#4997)

Co-authored-by: Yuta Imazu <yuta.imazu@linecorp.com>
Signed-off-by: Duo Zhang <zhangduo@apache.or
(cherry picked from commit ef6a1130d0f1a186b1d31f11be26ec95e3588ae2)
---
 .../regionserver/MetricsSource.java           |  6 +++-
 .../regionserver/ReplicationSource.java       |  9 +++--
 .../ReplicationSourceManager.java             |  3 +-
 .../replication/ReplicationSourceDummy.java   |  2 ++
 .../TestReplicationSourceManager.java         | 35 +++++++++++++++++++
 5 files changed, 50 insertions(+), 5 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsSource.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsSource.java
index 69671c13b61..f78eba3140c 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsSource.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsSource.java
@@ -272,10 +272,14 @@ public class MetricsSource implements BaseSource {
 
   /** Removes all metrics about this Source. */
   public void clear() {
+    terminate();
+    singleSourceSource.clear();
+  }
+
+  public void terminate() {
     int lastQueueSize = singleSourceSource.getSizeOfLogQueue();
     globalSourceSource.decrSizeOfLogQueue(lastQueueSize);
     singleSourceSource.decrSizeOfLogQueue(lastQueueSize);
-    singleSourceSource.clear();
     globalSourceSource.decrSizeOfHFileRefsQueue(lastHFileRefsQueueSize);
     lastShippedTimeStamps.clear();
     lastHFileRefsQueueSize = 0;
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 2db8acbd2dc..5fd000944bf 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
@@ -719,10 +719,13 @@ public class ReplicationSource implements ReplicationSourceInterface {
         }
       }
     }
-    if (clearMetrics) {
-      // Can be null in test context.
-      if (this.metrics != null) {
+
+    // Can be null in test context.
+    if (this.metrics != null) {
+      if (clearMetrics) {
         this.metrics.clear();
+      } else {
+        this.metrics.terminate();
       }
     }
   }
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 f32e62b749a..cbda89ede93 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
@@ -370,7 +370,8 @@ public class ReplicationSourceManager {
       ReplicationSourceInterface toRemove = this.sources.remove(peerId);
       if (toRemove != null) {
         LOG.info("Terminate replication source for " + toRemove.getPeerId());
-        toRemove.terminate(terminateMessage, null, true);
+        // Do not clear metrics
+        toRemove.terminate(terminateMessage, null, false);
       }
       src = createSource(peerId, peer);
       this.sources.put(peerId, src);
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 0ec2b0a8be8..ac553de178d 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
@@ -92,6 +92,8 @@ public class ReplicationSourceDummy implements ReplicationSourceInterface {
   public void terminate(String reason, Exception e, boolean clearMetrics) {
     if (clearMetrics) {
       this.metrics.clear();
+    } else {
+      this.metrics.terminate();
     }
   }
 
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java
index 550159eeb7f..268aa65ab18 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java
@@ -554,6 +554,41 @@ public abstract class TestReplicationSourceManager {
     }
   }
 
+  @Test
+  public void testDisablePeerMetricsCleanup() throws Exception {
+    final String peerId = "DummyPeer";
+    final ReplicationPeerConfig peerConfig = ReplicationPeerConfig.newBuilder()
+      .setClusterKey(utility.getZkCluster().getAddress().toString() + ":/hbase").build();
+    try {
+      MetricsReplicationSourceSource globalSource = getGlobalSource();
+      final int globalLogQueueSizeInitial = globalSource.getSizeOfLogQueue();
+      final long sizeOfLatestPath = getSizeOfLatestPath();
+      addPeerAndWait(peerId, peerConfig, true);
+      assertEquals(sizeOfLatestPath + globalLogQueueSizeInitial, globalSource.getSizeOfLogQueue());
+      ReplicationSourceInterface source = manager.getSource(peerId);
+      // Sanity check
+      assertNotNull(source);
+      final int sizeOfSingleLogQueue = source.getSourceMetrics().getSizeOfLogQueue();
+      // Enqueue log and check if metrics updated
+      source.enqueueLog(new Path("abc"));
+      assertEquals(1 + sizeOfSingleLogQueue, source.getSourceMetrics().getSizeOfLogQueue());
+      assertEquals(source.getSourceMetrics().getSizeOfLogQueue() + globalLogQueueSizeInitial,
+        globalSource.getSizeOfLogQueue());
+
+      // Refreshing the peer should decrement the global and single source metrics
+      manager.refreshSources(peerId);
+      assertEquals(globalLogQueueSizeInitial, globalSource.getSizeOfLogQueue());
+
+      source = manager.getSource(peerId);
+      assertNotNull(source);
+      assertEquals(sizeOfSingleLogQueue, source.getSourceMetrics().getSizeOfLogQueue());
+      assertEquals(source.getSourceMetrics().getSizeOfLogQueue() + globalLogQueueSizeInitial,
+        globalSource.getSizeOfLogQueue());
+    } finally {
+      removePeerAndWait(peerId);
+    }
+  }
+
   /**
    * Add a peer and wait for it to initialize
    * @param waitForSource Whether to wait for replication source to initialize