You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hbase.apache.org by "junegunn (via GitHub)" <gi...@apache.org> on 2026/09/10 11:52:18 UTC
[PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
junegunn opened a new pull request, #8646:
URL: https://github.com/apache/hbase/pull/8646
Jira: HBASE-30377
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.
```
startTime = 100, endTime = 200
WAL created at t=50, rolled at t=150 -> holds entries 50..150
50 <= 200 passes, but 50 >= 100 fails -> file skipped
lost: every entry in 100..150
```
### Fix
Compare against the modification time instead, which bounds the entries from above. That is only meaningful once the file is closed, since HDFS leaves mtime at the creation time through `hflush` and `hsync`, so it is gated on `DistributedFileSystem.isFileClosed`. Anything we cannot answer for, including non-HDFS filesystems, is treated as open and kept.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "mosmeh (via GitHub)" <gi...@apache.org>.
mosmeh commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4013196939
##########
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java:
##########
@@ -391,18 +393,44 @@ 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 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<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;
}
+ // 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())) {
Review Comment:
This can exclude WALs that the previous code retained, for example, when the NameNode clock (which determines mtime) lags behind the RegionServer clock (which determines the WAL filename timestamp).
How about applying this condition only when `timestamp < startTime`, so it only excludes WALs that the previous code would also have rejected?
```suggestion
if (timestamp < startTime && lfs.getModificationTime() < startTime && isClosed(fs, lfs.getPath())) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "junegunn (via GitHub)" <gi...@apache.org>.
junegunn commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4015067260
##########
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"));
Review Comment:
WAL files are named with millisecond timestamps, and `listLocatedStatus` iterates in name order on HDFS. This ordering has been stable since the test was written. The line is also pre-existing code, not part of this change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "junegunn (via GitHub)" <gi...@apache.org>.
junegunn commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4013588424
##########
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java:
##########
@@ -391,18 +393,44 @@ 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 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<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;
}
+ // 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())) {
Review Comment:
Ah, yeah, that's possible. Pretty unlikely in real-world use cases though, but it's better to be right.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "junegunn (via GitHub)" <gi...@apache.org>.
junegunn commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4013222256
##########
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java:
##########
@@ -391,18 +393,44 @@ 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 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<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;
}
+ // 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())) {
Review Comment:
Excellent point. Thank you very much.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4015540899
##########
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"));
Review Comment:
This test assumes a stable ordering of `splits` (i.e., that the WAL spanning `startTime` is always `splits.get(0)`). `InputFormat#getSplits()` ordering can vary depending on filesystem listing order, making this potentially flaky. Consider asserting the set of split contents regardless of order (e.g., run `testSplit` against both splits and verify both expected keys are present), or deterministically sort splits by path before indexing.
##########
hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALRecordReader.java:
##########
@@ -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);
Review Comment:
The updated comment refers to the last WAL's *modification time*, but the variable name `thirdTs` reads like a timestamp derived from the WAL name/creation time. If `thirdTs` is not actually the file mtime, the comment is misleading—either adjust the comment to match what `thirdTs` represents, or rename/compute the value to reflect modification time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "mosmeh (via GitHub)" <gi...@apache.org>.
mosmeh commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4013508372
##########
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java:
##########
@@ -391,18 +393,44 @@ 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 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<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;
}
+ // 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())) {
Review Comment:
Sorry for meat-proxying, but Codex says
> [P1] Refresh file status after confirming that the WAL is closed — hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java:372-372
> If a WAL closes after `listLocatedStatus` captures its status but before `isClosed` runs, `lfs` still contains the open file's creation-time mtime. This check then sees a closed file and incorrectly prunes it despite in-range entries and a newer actual mtime, silently omitting edits from WALPlayer. Re-fetch the file status after confirming closure before using its mtime to prune, retaining the file if that refresh fails.
With the `timestamp < startTime` guard, this patch still is an improvement, though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "junegunn (via GitHub)" <gi...@apache.org>.
junegunn commented on PR #8646:
URL: https://github.com/apache/hbase/pull/8646#issuecomment-5689824392
Copilot's test suggestions were not relevant, and I'm confident that this PR is now ready for review.
I also asked Codex to review the PR again and it commented:
> One existing limitation remains: `hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java:414` compares NameNode modification time against RegionServer entry timestamps. If the RegionServer clock is ahead, a closed, unarchived WAL can still lose in-range entries during filtering. This case also failed before the PR.
While this is technically possible, I wouldn't worry too much about it in real-world HBase clusters, where NTP is a requirement. More importantly, this is an existing limitation and not a regression introduced by the PR.
Could someone take a look at the patch? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "junegunn (via GitHub)" <gi...@apache.org>.
junegunn commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4013757273
##########
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java:
##########
@@ -391,18 +393,44 @@ 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 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<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;
}
+ // 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())) {
Review Comment:
Addressed in 5adb2557f76905aadba3dd7e99dea1996e57dc26. Could you take another look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "junegunn (via GitHub)" <gi...@apache.org>.
junegunn merged PR #8646:
URL: https://github.com/apache/hbase/pull/8646
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "junegunn (via GitHub)" <gi...@apache.org>.
junegunn commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4015069947
##########
hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALInputFormat.java:
##########
@@ -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);
+ }
Review Comment:
`isKept` is a `private static` helper that does not declare `throws`. `Mockito.when(...).thenReturn(...)` requires catching the checked `IOException` inside the helper itself. Pre-existing code, not part of this change.
##########
hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALRecordReader.java:
##########
@@ -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);
Review Comment:
`thirdTs` is sampled immediately after `log.shutdown()`, which flushes and closes the WAL. The WAL's final modification time is set at close, so `thirdTs >= mtime` holds. Adding `+ 1` makes it strictly greater. The comment describes the relationship correctly. The variable name is pre-existing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "junegunn (via GitHub)" <gi...@apache.org>.
junegunn commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4016374174
##########
hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALRecordReader.java:
##########
@@ -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);
Review Comment:
Already answered above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4014834591
##########
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"));
Review Comment:
This test now assumes a deterministic ordering of `splits` (i.e., that the split containing entry \"2\" is always index 0 and \"3\" is always index 1). `InputFormat#getSplits` ordering can vary by filesystem iteration order, which can make this test flaky. Prefer asserting independent of order (e.g., locate the split by its WAL path/name, or sort splits by path before calling `testSplit`).
##########
hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALInputFormat.java:
##########
@@ -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);
+ }
Review Comment:
This helper wraps `IOException` in a `RuntimeException` even though the calling test already declares `throws Exception`. Simplifying this to let the exception propagate (or moving the stubbing into the calling test method that already throws) will make failures easier to diagnose and reduce unnecessary control flow.
##########
hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALRecordReader.java:
##########
@@ -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);
Review Comment:
The comment states `START_TIME_KEY` is set strictly after the last WAL's *modification time*, but the variable name `thirdTs` reads like a timestamp from the WAL name/creation time. If `thirdTs` is not actually the last WAL mtime, the comment is misleading and the test may be relying on an incorrect assumption. Consider either (a) renaming/deriving the value from the actual file modification time used by the logic under test, or (b) updating the comment to match what `thirdTs` represents.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] HBASE-30377 WALInputFormat drops WAL files that span the requested time range [hbase]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #8646:
URL: https://github.com/apache/hbase/pull/8646#discussion_r4015194278
##########
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALInputFormat.java:
##########
@@ -391,18 +393,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.
Review Comment:
The `getFiles` Javadoc above still says that a timestamped file is filtered whenever its name is older than `startTime`, but this new branch intentionally retains such files when they may still be open. Please update that public behavior description so callers and future maintainers do not rely on the now-incorrect name-only filtering contract.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hbase.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org