HBASE-25924 Re-compute size of WAL file while removing from WALEntryStream (#3316)

Signed-off-by: Andrew Purtell <apurtell@apache.org>

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsReplicationSourceSource.java

@@ -70,6 +70,7 @@ public interface MetricsReplicationSourceSource extends BaseSource {
   void decrSizeOfHFileRefsQueue(long size);
   void incrUnknownFileLengthForClosedWAL();
   void incrUncleanlyClosedWALs();
+  long getUncleanlyClosedWALs();
   void incrBytesSkippedInUncleanlyClosedWALs(final long bytes);
   void incrRestartedWALReading();
   void incrRepeatedFileBytes(final long bytes);

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsReplicationGlobalSourceSource.java

@@ -175,6 +175,12 @@ public class MetricsReplicationGlobalSourceSource implements MetricsReplicationS
   public void incrUncleanlyClosedWALs() {
     uncleanlyClosedWAL.incr(1L);
   }
+
+  @Override
+  public long getUncleanlyClosedWALs() {
+    return uncleanlyClosedWAL.value();
+  }
+
   @Override
   public void incrBytesSkippedInUncleanlyClosedWALs(final long bytes) {
     uncleanlyClosedSkippedBytes.incr(bytes);

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsReplicationSourceSourceImpl.java

@@ -241,6 +241,11 @@ public class MetricsReplicationSourceSourceImpl implements MetricsReplicationSou
     uncleanlyClosedWAL.incr(1L);
   }
 
+  @Override
+  public long getUncleanlyClosedWALs() {
+    return uncleanlyClosedWAL.value();
+  }
+
   @Override
   public void incrBytesSkippedInUncleanlyClosedWALs(final long bytes) {
     uncleanlyClosedSkippedBytes.incr(bytes);

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/ProtobufLogReader.java

@@ -203,6 +203,10 @@ public class ProtobufLogReader extends ReaderBase {
   private String initInternal(FSDataInputStream stream, boolean isFirst)
       throws IOException {
     close();
+    if (!isFirst) {
+      // Re-compute the file length.
+      this.fileLength = fs.getFileStatus(path).getLen();
+    }
     long expectedPos = PB_WAL_MAGIC.length;
     if (stream == null) {
       stream = fs.open(path);

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

@@ -306,6 +306,15 @@ public class MetricsSource implements BaseSource {
     return lastTimestamp;
   }
 
+
+  /**
+   * Get the value of uncleanlyClosedWAL counter
+   * @return uncleanlyClosedWAL
+   */
+  public long getUncleanlyClosedWALs() {
+    return singleSourceSource.getUncleanlyClosedWALs();
+  }
+
   /**
    * Get the source initializing counts
    * @return number of replication sources getting initialized

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

@@ -899,4 +899,22 @@ public class TestWALEntryStream {
     // After removing one wal, size of log queue will be 1 again.
     assertEquals(1, logQueue.getMetrics().getSizeOfLogQueue());
   }
+
+  /**
+   * Tests that wals are closed cleanly and we read the trailer when we remove wal
+   * from WALEntryStream.
+   */
+  @Test
+  public void testCleanClosedWALs() throws Exception {
+    try (WALEntryStream entryStream = new WALEntryStream(
+      logQueue, fs, conf, logQueue.getMetrics(), fakeWalGroupId)) {
+      assertEquals(0, logQueue.getMetrics().getUncleanlyClosedWALs());
+      appendToLog();
+      assertNotNull(entryStream.next());
+      log.rollWriter();
+      appendToLog();
+      assertNotNull(entryStream.next());
+      assertEquals(0, logQueue.getMetrics().getUncleanlyClosedWALs());
+    }
+  }
 }