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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
