[hbase] branch branch-2.4 updated: HBASE-25679 Size of log queue metric is incorrect (#3071)

[st...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

stack pushed a commit to branch branch-2.4
in repository https://gitbox.apache.org/repos/asf/hbase.git


The following commit(s) were added to refs/heads/branch-2.4 by this push:
     new 27e9c28  HBASE-25679 Size of log queue metric is incorrect (#3071)
27e9c28 is described below

commit 27e9c281da6c594ab5481d7c3a73cc29f7f19992
Author: shahrs87 <sh...@gmail.com>
AuthorDate: Fri Mar 19 16:14:45 2021 -0700

    HBASE-25679 Size of log queue metric is incorrect (#3071)
    
    Co-authored-by: Rushabh <ru...@salesforce.com>
    Signed-off-by: stack <st...@apache.org>
---
 .../regionserver/ReplicationSourceLogQueue.java    |  4 +++
 .../replication/regionserver/WALEntryStream.java   |  1 -
 .../regionserver/TestWALEntryStream.java           | 30 +++++++++++++++++++++-
 3 files changed, 33 insertions(+), 2 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceLogQueue.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceLogQueue.java
index 8a774fb..4d89ede 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceLogQueue.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceLogQueue.java
@@ -186,4 +186,8 @@ public class ReplicationSourceLogQueue {
     }
     return oldestWalTimestamp;
   }
+
+  public MetricsSource getMetrics() {
+    return metrics;
+  }
 }
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/WALEntryStream.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/WALEntryStream.java
index 721a122..671d3a9 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/WALEntryStream.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/WALEntryStream.java
@@ -256,7 +256,6 @@ class WALEntryStream implements Closeable {
     logQueue.remove(walGroupId);
     setCurrentPath(null);
     setPosition(0);
-    metrics.decrSizeOfLogQueue();
   }
 
   /**
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestWALEntryStream.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestWALEntryStream.java
index d31b864..ae9bb67 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestWALEntryStream.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestWALEntryStream.java
@@ -135,7 +135,10 @@ public class TestWALEntryStream {
   @Before
   public void setUp() throws Exception {
     ReplicationSource source = mock(ReplicationSource.class);
-    logQueue = new ReplicationSourceLogQueue(CONF, new MetricsSource("2"), source);
+    MetricsSource metricsSource = new MetricsSource("2");
+    // Source with the same id is shared and carries values from the last run
+    metricsSource.clear();
+    logQueue = new ReplicationSourceLogQueue(CONF, metricsSource, source);
     pathWatcher = new PathWatcher();
     final WALFactory wals = new WALFactory(CONF, tn.getMethodName());
     wals.getWALProvider().addWALActionsListener(pathWatcher);
@@ -766,4 +769,29 @@ public class TestWALEntryStream {
     }
     writer.close();
   }
+
+  /**
+   * Tests size of log queue is incremented and decremented properly.
+   */
+  @Test
+  public void testSizeOfLogQueue() throws Exception {
+    // There should be always 1 log which is current wal.
+    assertEquals(1, logQueue.getMetrics().getSizeOfLogQueue());
+    appendToLogAndSync();
+
+    log.rollWriter();
+    // After rolling there will be 2 wals in the queue
+    assertEquals(2, logQueue.getMetrics().getSizeOfLogQueue());
+
+    try (WALEntryStream entryStream = new WALEntryStream(
+      logQueue, CONF, 0, log, null, logQueue.getMetrics(), fakeWalGroupId)) {
+      // There's one edit in the log, read it.
+      assertTrue(entryStream.hasNext());
+      WAL.Entry entry = entryStream.next();
+      assertNotNull(entry);
+      assertFalse(entryStream.hasNext());
+    }
+    // After removing one wal, size of log queue will be 1 again.
+    assertEquals(1, logQueue.getMetrics().getSizeOfLogQueue());
+  }
 }