From 016d91c828e1dccfb6f79e0e26f7b4c952a1ce95 Mon Sep 17 00:00:00 2001 From: Junegunn Choi Date: Thu, 10 Sep 2026 20:49:25 +0900 Subject: [PATCH 1/6] HBASE-30377 WALInputFormat drops WAL files that span the requested time range The timestamp in a WAL's name is 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, and comparing the name against startTime threw the whole file away. Compare against the modification time instead, which bounds the entries from above, but only once the file is closed: HDFS leaves mtime at the creation time through hflush and hsync. Gate it on DistributedFileSystem.isFileClosed and keep anything we cannot answer for, including non-HDFS filesystems. The call is short-circuited. wal.start.time defaults to Long.MIN_VALUE, so a job that does not ask for a start time never reaches it, and one that does pays it only for files mtime alone would prune. TestWALRecordReader.testPartialRead asserted the old behaviour: it writes an entry at exactly startTime into a WAL created earlier and expected that file to be skipped. Corrected to expect both splits, asserting the entry that was being lost. --- .../hbase/mapreduce/WALInputFormat.java | 46 +++++++++--- .../hbase/mapreduce/TestWALInputFormat.java | 71 +++++++++++++++---- .../hbase/mapreduce/TestWALRecordReader.java | 7 +- 3 files changed, 98 insertions(+), 26 deletions(-) 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..bc4e6f534f3a 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; @@ -382,7 +384,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 +393,44 @@ 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 known to be closed. Only a closed file has a final modification time, so + * only then can it be used as an upper bound on the entries inside. Anything we cannot answer + * for, including non-HDFS filesystems, is reported as open so that the file is kept. + */ + private static boolean isClosed(FileSystem fs, Path path) { + try { + FileSystem backing = fs instanceof HFileSystem ? ((HFileSystem) fs).getBackingFs() : fs; + return backing instanceof DistributedFileSystem + && ((DistributedFileSystem) backing).isFileClosed(path); + } catch (IOException e) { + LOG.debug("Could not tell whether {} is closed, keeping it", path, 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; } + // The modification time is the upper bound, but HDFS leaves it at the creation time until + // the file is closed, so it is only meaningful once the file is. Order the checks so the + // extra RPC is only paid for files that the modification time alone would prune. + if (lfs.getModificationTime() < startTime && isClosed(fs, lfs.getPath())) { + 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..d9a3443362a1 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,74 @@ 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) { + List result = new ArrayList<>(); + LocatedFileStatus lfs = Mockito.mock(LocatedFileStatus.class); + Mockito.when(lfs.getPath()).thenReturn(new Path("/name." + created)); + Mockito.when(lfs.getModificationTime()).thenReturn(mtime); + 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)); } @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..dcfea74cf044 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")); } /** From 7d4588ab4af58b4cc40d4f2350d7a7d4ed85b8ae Mon Sep 17 00:00:00 2001 From: Junegunn Choi Date: Tue, 15 Sep 2026 16:45:45 +0900 Subject: [PATCH 2/6] Guard mtime-based pruning with timestamp < startTime Co-authored-by: Yuta Imazu --- .../java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 bc4e6f534f3a..6ea26468105b 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 @@ -424,7 +424,7 @@ static void addFile(List result, FileSystem fs, LocatedFileStatus lf // The modification time is the upper bound, but HDFS leaves it at the creation time until // the file is closed, so it is only meaningful once the file is. Order the checks so the // extra RPC is only paid for files that the modification time alone would prune. - if (lfs.getModificationTime() < startTime && isClosed(fs, lfs.getPath())) { + if (timestamp < startTime && lfs.getModificationTime() < startTime && isClosed(fs, lfs.getPath())) { LOG.info("Skipped {}, closed before startTime [{}/{}]", lfs.getPath(), startTime, Instant.ofEpochMilli(startTime)); return; From 5adb2557f76905aadba3dd7e99dea1996e57dc26 Mon Sep 17 00:00:00 2001 From: Junegunn Choi Date: Tue, 15 Sep 2026 17:45:42 +0900 Subject: [PATCH 3/6] Re-fetch file status after confirming WAL closure listLocatedStatus captures mtime while the file is open (mtime = creation time). If the file closes before isFileClosed runs, the stale mtime can cause incorrect pruning. Re-fetch FileStatus after confirming closure to get the final mtime. - Rename isClosed to isClosedBefore, fold mtime check and re-fetch - Stale mtime check remains as fast path to skip the RPC --- .../hbase/mapreduce/WALInputFormat.java | 30 ++++++++++++------- .../hbase/mapreduce/TestWALInputFormat.java | 21 +++++++++++-- 2 files changed, 38 insertions(+), 13 deletions(-) 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 6ea26468105b..0517c8abdc32 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 @@ -394,17 +394,28 @@ List getFiles(FileSystem fs, Path dir, long startTime, long endTime, } /** - * Whether the file is known to be closed. Only a closed file has a final modification time, so - * only then can it be used as an upper bound on the entries inside. Anything we cannot answer - * for, including non-HDFS filesystems, is reported as open so that the file is kept. + * 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 isClosed(FileSystem fs, Path path) { + 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; - return backing instanceof DistributedFileSystem - && ((DistributedFileSystem) backing).isFileClosed(path); + if ( + !(backing instanceof DistributedFileSystem) + || !((DistributedFileSystem) backing).isFileClosed(lfs.getPath()) + ) { + return false; + } + FileStatus refreshed = fs.getFileStatus(lfs.getPath()); + return refreshed.getModificationTime() < time; } catch (IOException e) { - LOG.debug("Could not tell whether {} is closed, keeping it", path, e); + LOG.debug("Could not confirm closure of {}, keeping it", lfs.getPath(), e); return false; } } @@ -421,10 +432,7 @@ static void addFile(List result, FileSystem fs, LocatedFileStatus lf Instant.ofEpochMilli(endTime)); return; } - // The modification time is the upper bound, but HDFS leaves it at the creation time until - // the file is closed, so it is only meaningful once the file is. Order the checks so the - // extra RPC is only paid for files that the modification time alone would prune. - if (timestamp < startTime && lfs.getModificationTime() < startTime && isClosed(fs, lfs.getPath())) { + if (timestamp < startTime && isClosedBefore(fs, lfs, startTime)) { LOG.info("Skipped {}, closed before startTime [{}/{}]", lfs.getPath(), startTime, Instant.ofEpochMilli(startTime)); return; 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 d9a3443362a1..f9ae74996d4d 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 @@ -100,10 +100,23 @@ public void testAddFile() { } 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(new Path("/name." + created)); - Mockito.when(lfs.getModificationTime()).thenReturn(mtime); + 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(); } @@ -134,6 +147,10 @@ public void testAddFileUsesModificationTimeOfClosedFilesOnly() throws Exception // 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)); } @Test From 55837b30202edb3a8da6102fded303066646fc20 Mon Sep 17 00:00:00 2001 From: Junegunn Choi Date: Tue, 15 Sep 2026 19:39:50 +0900 Subject: [PATCH 4/6] Catch UnsupportedOperationException from isFileClosed ViewDistributedFileSystem passes the instanceof DistributedFileSystem check but throws UnsupportedOperationException when the mounted filesystem does not support isFileClosed(). Treat this the same as a non-HDFS filesystem: keep the file. --- .../org/apache/hadoop/hbase/mapreduce/WALInputFormat.java | 2 +- .../apache/hadoop/hbase/mapreduce/TestWALInputFormat.java | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) 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 0517c8abdc32..d2409ca81223 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 @@ -414,7 +414,7 @@ private static boolean isClosedBefore(FileSystem fs, LocatedFileStatus lfs, long } FileStatus refreshed = fs.getFileStatus(lfs.getPath()); return refreshed.getModificationTime() < time; - } catch (IOException e) { + } catch (IOException | UnsupportedOperationException e) { LOG.debug("Could not confirm closure of {}, keeping it", lfs.getPath(), e); return false; } 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 f9ae74996d4d..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 @@ -151,6 +151,13 @@ public void testAddFileUsesModificationTimeOfClosedFilesOnly() throws Exception // 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 From 8483fe0da42d5d8b71cb72a5718804ba25a112b8 Mon Sep 17 00:00:00 2001 From: Junegunn Choi Date: Tue, 15 Sep 2026 19:40:11 +0900 Subject: [PATCH 5/6] Use thirdTs + 1 as startTime in testWALRecordReader thirdTs is sampled immediately after WAL shutdown, so it can equal the WAL's final modification time. Using it directly as startTime makes isClosedBefore retain the WAL, failing the assertTrue(splits.isEmpty()) assertion. Offset by 1 ms to make the boundary unambiguous. --- .../apache/hadoop/hbase/mapreduce/TestWALRecordReader.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 dcfea74cf044..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 @@ -248,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()); } From 66583d89baa6c0c570b330ca7471f37889b65371 Mon Sep 17 00:00:00 2001 From: Junegunn Choi Date: Tue, 15 Sep 2026 20:50:21 +0900 Subject: [PATCH 6/6] Update getFiles Javadoc to reflect closure-aware filtering The old description implied pure name-based filtering, but files created before startTime are only dropped when confirmed closed. --- .../apache/hadoop/hbase/mapreduce/WALInputFormat.java | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) 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 d2409ca81223..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 @@ -363,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 {