HBASE-30377 WALInputFormat drops WAL files that span the requested time range

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java

@@ -32,6 +32,7 @@
 import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.hbase.fs.HFileSystem;
 import org.apache.hadoop.hbase.regionserver.wal.WALHeaderEOFException;
 import org.apache.hadoop.hbase.util.LeaseNotRecoveredException;
 import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
@@ -41,6 +42,7 @@
 import org.apache.hadoop.hbase.wal.WALFactory;
 import org.apache.hadoop.hbase.wal.WALKey;
 import org.apache.hadoop.hbase.wal.WALStreamReader;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.mapreduce.InputFormat;
 import org.apache.hadoop.mapreduce.InputSplit;
@@ -361,12 +363,10 @@ Path[] getInputPaths(Configuration conf) {
   }
 
   /**
-   * @param startTime If file looks like it has a timestamp in its name, we'll check if newer or
-   *                  equal to this value else we will filter out the file. If name does not seem to
-   *                  have a timestamp, we will just return it w/o filtering.
-   * @param endTime   If file looks like it has a timestamp in its name, we'll check if older or
-   *                  equal to this value else we will filter out the file. If name does not seem to
-   *                  have a timestamp, we will just return it w/o filtering.
+   * @param startTime Files created before this time are dropped only if confirmed closed before it.
+   *                  Files without a parseable timestamp in their name are always included.
+   * @param endTime   Files created after this time are dropped. Files without a parseable timestamp
+   *                  in their name are always included.
    */
   List<FileStatus> getFiles(FileSystem fs, Path dir, long startTime, long endTime,
     Configuration conf) throws IOException {
@@ -382,7 +382,7 @@ List<FileStatus> getFiles(FileSystem fs, Path dir, long startTime, long endTime,
         // Recurse into sub directories
         result.addAll(getFiles(fs, file.getPath(), startTime, endTime, conf));
       } else {
-        addFile(result, file, startTime, endTime);
+        addFile(result, fs, file, startTime, endTime);
       }
     }
     // TODO: These results should be sorted? Results could be content of recovered.edits directory
@@ -391,18 +391,52 @@ List<FileStatus> getFiles(FileSystem fs, Path dir, long startTime, long endTime,
     return result;
   }
 
-  static void addFile(List<FileStatus> result, LocatedFileStatus lfs, long startTime,
+  /**
+   * Whether the file is closed and its final modification time precedes {@code time}. Only a closed
+   * file has a reliable modification time, so an open file or a non-HDFS file always returns
+   * {@code false} (kept). When the file is confirmed closed, its status is re-fetched because the
+   * {@code lfs} from {@code listLocatedStatus} may carry a stale creation-time mtime from when the
+   * file was still open.
+   */
+  private static boolean isClosedBefore(FileSystem fs, LocatedFileStatus lfs, long time) {
+    if (lfs.getModificationTime() >= time) {
+      return false;
+    }
+    try {
+      FileSystem backing = fs instanceof HFileSystem ? ((HFileSystem) fs).getBackingFs() : fs;
+      if (
+        !(backing instanceof DistributedFileSystem)
+          || !((DistributedFileSystem) backing).isFileClosed(lfs.getPath())
+      ) {
+        return false;
+      }
+      FileStatus refreshed = fs.getFileStatus(lfs.getPath());
+      return refreshed.getModificationTime() < time;
+    } catch (IOException | UnsupportedOperationException e) {
+      LOG.debug("Could not confirm closure of {}, keeping it", lfs.getPath(), e);
+      return false;
+    }
+  }
+
+  static void addFile(List<FileStatus> result, FileSystem fs, LocatedFileStatus lfs, long startTime,
     long endTime) {
     long timestamp = AbstractFSWALProvider.getTimestamp(lfs.getPath().getName());
     if (timestamp > 0) {
-      // Looks like a valid timestamp.
-      if (timestamp <= endTime && timestamp >= startTime) {
-        LOG.info("Found {}", lfs.getPath());
-        result.add(lfs);
-      } else {
-        LOG.info("Skipped {}, outside range [{}/{} - {}/{}]", lfs.getPath(), startTime,
-          Instant.ofEpochMilli(startTime), endTime, Instant.ofEpochMilli(endTime));
+      // The name carries the WAL's creation time, which only bounds its entries from below. A WAL
+      // stays open until it rolls, so one created before startTime can still hold entries in
+      // range and must not be dropped on the strength of its name alone.
+      if (timestamp > endTime) {
+        LOG.info("Skipped {}, created after endTime [{}/{}]", lfs.getPath(), endTime,
+          Instant.ofEpochMilli(endTime));
+        return;
+      }
+      if (timestamp < startTime && isClosedBefore(fs, lfs, startTime)) {
+        LOG.info("Skipped {}, closed before startTime [{}/{}]", lfs.getPath(), startTime,
+          Instant.ofEpochMilli(startTime));
+        return;
       }
+      LOG.info("Found {}", lfs.getPath());
+      result.add(lfs);
     } else {
       // If no timestamp, add it regardless.
       LOG.info("Found (no-timestamp!) {}", lfs);

hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALInputFormat.java

@@ -38,6 +38,7 @@
 import org.apache.hadoop.hbase.testclassification.MediumTests;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapreduce.JobContext;
@@ -65,34 +66,98 @@ public static void setupClass() throws Exception {
   @Test
   public void testAddFile() {
     List<FileStatus> lfss = new ArrayList<>();
+    // a plain FileSystem is never reported closed, so nothing is skipped on the startTime side
+    FileSystem fs = Mockito.mock(FileSystem.class);
     LocatedFileStatus lfs = Mockito.mock(LocatedFileStatus.class);
     long now = EnvironmentEdgeManager.currentTime();
     Mockito.when(lfs.getPath()).thenReturn(new Path("/name." + now));
-    WALInputFormat.addFile(lfss, lfs, now, now);
+    WALInputFormat.addFile(lfss, fs, lfs, now, now);
     assertEquals(1, lfss.size());
-    WALInputFormat.addFile(lfss, lfs, now - 1, now - 1);
+    WALInputFormat.addFile(lfss, fs, lfs, now - 1, now - 1);
     assertEquals(1, lfss.size());
-    WALInputFormat.addFile(lfss, lfs, now - 2, now - 1);
+    WALInputFormat.addFile(lfss, fs, lfs, now - 2, now - 1);
     assertEquals(1, lfss.size());
-    WALInputFormat.addFile(lfss, lfs, now - 2, now);
+    WALInputFormat.addFile(lfss, fs, lfs, now - 2, now);
     assertEquals(2, lfss.size());
-    WALInputFormat.addFile(lfss, lfs, Long.MIN_VALUE, now);
+    WALInputFormat.addFile(lfss, fs, lfs, Long.MIN_VALUE, now);
     assertEquals(3, lfss.size());
-    WALInputFormat.addFile(lfss, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
+    WALInputFormat.addFile(lfss, fs, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
     assertEquals(4, lfss.size());
-    WALInputFormat.addFile(lfss, lfs, now, now + 2);
+    WALInputFormat.addFile(lfss, fs, lfs, now, now + 2);
     assertEquals(5, lfss.size());
-    WALInputFormat.addFile(lfss, lfs, now + 1, now + 2);
-    assertEquals(5, lfss.size());
-    Mockito.when(lfs.getPath()).thenReturn(new Path("/name"));
-    WALInputFormat.addFile(lfss, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
+    // created before startTime, but it may have stayed open and collected in-range entries
+    WALInputFormat.addFile(lfss, fs, lfs, now + 1, now + 2);
     assertEquals(6, lfss.size());
-    Mockito.when(lfs.getPath()).thenReturn(new Path("/name.123"));
-    WALInputFormat.addFile(lfss, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
+    Mockito.when(lfs.getPath()).thenReturn(new Path("/name"));
+    WALInputFormat.addFile(lfss, fs, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
     assertEquals(7, lfss.size());
-    Mockito.when(lfs.getPath()).thenReturn(new Path("/name." + now + ".meta"));
-    WALInputFormat.addFile(lfss, lfs, now, now);
+    Mockito.when(lfs.getPath()).thenReturn(new Path("/name.123"));
+    WALInputFormat.addFile(lfss, fs, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
     assertEquals(8, lfss.size());
+    Mockito.when(lfs.getPath()).thenReturn(new Path("/name." + now + ".meta"));
+    WALInputFormat.addFile(lfss, fs, lfs, now, now);
+    assertEquals(9, lfss.size());
+  }
+
+  private static boolean isKept(FileSystem fs, long created, long mtime, long start, long end) {
+    return isKept(fs, created, mtime, mtime, start, end);
+  }
+
+  private static boolean isKept(FileSystem fs, long created, long staleMtime, long refreshedMtime,
+    long start, long end) {
+    List<FileStatus> result = new ArrayList<>();
+    Path path = new Path("/name." + created);
+    LocatedFileStatus lfs = Mockito.mock(LocatedFileStatus.class);
+    Mockito.when(lfs.getPath()).thenReturn(path);
+    Mockito.when(lfs.getModificationTime()).thenReturn(staleMtime);
+    try {
+      FileStatus refreshed = Mockito.mock(FileStatus.class);
+      Mockito.when(refreshed.getModificationTime()).thenReturn(refreshedMtime);
+      Mockito.when(fs.getFileStatus(path)).thenReturn(refreshed);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+    WALInputFormat.addFile(result, fs, lfs, start, end);
+    return !result.isEmpty();
+  }
+
+  /**
+   * The name of a WAL carries its creation time, which only bounds its entries from below. A WAL
+   * stays open until it rolls, so one created before startTime can still hold entries in range.
+   * Only a closed file has a final modification time that can rule it out.
+   */
+  @Test
+  public void testAddFileUsesModificationTimeOfClosedFilesOnly() throws Exception {
+    long now = EnvironmentEdgeManager.currentTime();
+
+    DistributedFileSystem closed = Mockito.mock(DistributedFileSystem.class);
+    Mockito.when(closed.isFileClosed(Mockito.any())).thenReturn(true);
+    DistributedFileSystem open = Mockito.mock(DistributedFileSystem.class);
+    Mockito.when(open.isFileClosed(Mockito.any())).thenReturn(false);
+
+    // Closed, and its last write predates the window: nothing in it can be in range.
+    assertFalse(isKept(closed, now - 100, now - 50, now, now + 100));
+
+    // Same file, but the window opens before it was closed, so it spans the boundary.
+    assertTrue(isKept(closed, now - 100, now - 50, now - 60, now + 100));
+
+    // Still open. Its mtime is stuck near the creation time and says nothing about the entries
+    // it may yet receive, so it has to be kept even though mtime is before the window.
+    assertTrue(isKept(open, now - 100, now - 100, now, now + 100));
+
+    // Created after the window closed: every entry in it is later still.
+    assertFalse(isKept(closed, now + 200, now + 200, now, now + 100));
+
+    // Race condition: file closed between listLocatedStatus and isFileClosed. The stale mtime
+    // from the listing predates the window, but the refreshed mtime (after close) does not.
+    assertTrue(isKept(closed, now - 100, now - 50, now + 10, now, now + 100));
+
+    // ViewDistributedFileSystem wrapping non-HDFS storage: isFileClosed throws
+    // UnsupportedOperationException. The file must be kept (same as non-HDFS).
+    DistributedFileSystem unsupported = Mockito.mock(DistributedFileSystem.class);
+    Mockito.when(unsupported.isFileClosed(Mockito.any()))
+      .thenThrow(new UnsupportedOperationException("mounted fs"));
+    assertTrue(isKept(unsupported, now - 100, now - 50, now, now + 100));
   }
 
   @Test

hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALRecordReader.java

@@ -190,9 +190,12 @@ public void testPartialRead() throws Exception {
     jobConf.setLong(WALInputFormat.START_TIME_KEY, ts + 1);
     jobConf.setLong(WALInputFormat.END_TIME_KEY, ts1 + 1);
     splits = input.getSplits(MapreduceTestingShim.createJobContext(jobConf));
-    assertEquals(1, splits.size());
+    assertEquals(2, splits.size());
+    // The 1st file was created before startTime but stayed open until it rolled, so its 2nd
+    // entry, written at exactly startTime, is in-range.
+    testSplit(splits.get(0), Bytes.toBytes("2"));
     // Only the 1st entry from the 2nd file is in-range.
-    testSplit(splits.get(0), Bytes.toBytes("3"));
+    testSplit(splits.get(1), Bytes.toBytes("3"));
   }
 
   /**
@@ -245,9 +248,9 @@ public void testWALRecordReader() throws Exception {
     assertEquals(1, splits.size());
     testSplit(splits.get(0), Bytes.toBytes("1"));
 
-    // now set a start time
+    // now set a start time strictly after the last WAL's modification time
     jobConf.setLong(WALInputFormat.END_TIME_KEY, Long.MAX_VALUE);
-    jobConf.setLong(WALInputFormat.START_TIME_KEY, thirdTs);
+    jobConf.setLong(WALInputFormat.START_TIME_KEY, thirdTs + 1);
     splits = input.getSplits(MapreduceTestingShim.createJobContext(jobConf));
     assertTrue(splits.isEmpty());
   }