HBASE-25679 Size of log queue metric is incorrect

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

@@ -181,4 +181,8 @@ private long getOldestWalTimestamp() {
     }
     return oldestWalTimestamp;
   }
+
+  public MetricsSource getMetrics() {
+    return metrics;
+  }
 }

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

@@ -271,7 +271,6 @@ private void dequeueCurrentLog() throws IOException {
     logQueue.remove(walGroupId);
     setCurrentPath(null);
     setPosition(0);
-    metrics.decrSizeOfLogQueue();
   }
 
   private void readNextEntryAndSetPosition() throws IOException {

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

@@ -143,7 +143,10 @@ public static void tearDownAfterClass() throws Exception {
 
   @Before
   public void setUp() throws Exception {
-    logQueue = new ReplicationSourceLogQueue(conf, new MetricsSource("2"));
+    MetricsSource source = new MetricsSource("2");
+    // Source with the same id is shared and carries values from the last run
+    source.clear();
+    logQueue = new ReplicationSourceLogQueue(conf, source);
     List<WALActionsListener> listeners = new ArrayList<WALActionsListener>();
     pathWatcher = new PathWatcher();
     listeners.add(pathWatcher);
@@ -811,4 +814,28 @@ private void appendEntries(WALProvider.Writer writer, int numEntries) throws IOE
     }
     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());
+    appendToLog();
+    log.rollWriter();
+    // After rolling there will be 2 wals in the queue
+    assertEquals(2, logQueue.getMetrics().getSizeOfLogQueue());
+
+    try (WALEntryStream entryStream =
+           new WALEntryStream(logQueue, fs, conf, 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());
+  }
 }