diff --git a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java b/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java index 6d1ea7a086ec..0146ff532d42 100644 --- a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java +++ b/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 getFiles(FileSystem fs, Path dir, long startTime, long endTime, Configuration conf) throws IOException { @@ -382,7 +382,7 @@ List 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 getFiles(FileSystem fs, Path dir, long startTime, long endTime, return result; } - static void addFile(List 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 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); diff --git a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALInputFormat.java b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALInputFormat.java index 9ece6a826eae..5fbaf9e21e76 100644 --- a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALInputFormat.java +++ b/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 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 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 diff --git a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALRecordReader.java b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALRecordReader.java index f72348ac3fe9..21edb41e2981 100644 --- a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALRecordReader.java +++ b/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()); }