[
https://issues.apache.org/jira/browse/HADOOP-17611?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18130798#comment-18130798
]
ASF GitHub Bot commented on HADOOP-17611:
-----------------------------------------
Copilot commented on code in PR #8858:
URL: https://github.com/apache/hadoop/pull/8858#discussion_r4358034303
##########
hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java:
##########
@@ -654,6 +654,13 @@ private void concatFileChunks(Configuration conf, Path
sourceFile,
DistCpUtils.compareFileLengthsAndChecksums(srcFileStatus.getLen(),
srcfs, sourceFile, null, dstfs,
targetFile, skipCrc, srcFileStatus.getLen());
+ // Concat changes the modification time preserved on the first chunk.
+ // Restore times after validation, which may also update the access time.
+ if (DistCpUtils.unpackAttributes(conf.get(CONF_LABEL_PRESERVE_STATUS))
+ .contains(FileAttribute.TIMES)) {
+ dstfs.setTimes(targetFile, srcFileStatus.getModificationTime(),
+ srcFileStatus.getAccessTime());
Review Comment:
`setTimes(Path, mtime, atime)` typically uses `-1` to mean 'do not change'.
If `srcFileStatus.getAccessTime()` is `0` (common on filesystems that don’t
expose atime), this will actively set the destination atime to epoch rather
than preserving/ignoring it. Consider passing `-1` when the source atime is
non-positive (e.g., compute `long atime = srcFileStatus.getAccessTime() > 0 ?
srcFileStatus.getAccessTime() : -1`) so enabling `TIMES` doesn’t inadvertently
clobber atime on FSes without atime support.
##########
hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/AbstractContractDistCpTest.java:
##########
@@ -252,6 +271,243 @@ protected Path getRemoteDir() {
return remoteDir;
}
+ @ParameterizedTest(name = "preserveTimes={0}, direct={1}, atomic={2}")
+ @CsvSource({
+ "true, false, false",
+ "false, false, false",
+ "true, true, false",
+ "false, true, false",
+ "true, false, true",
+ "false, false, true"
+ })
+ public void testChunkCopyPreserveTimes(boolean preserveTimes,
+ boolean directWrite, boolean atomicCommit) throws Exception {
+ requireChunkCopySupport();
+ Path source = new Path(remoteDir, "source");
+ Path target = new Path(remoteDir, "target");
+ Map<String, FileStatus> sourceStatuses = createChunkCopyTree(source);
+
+ DistCpOptions.Builder builder = chunkCopyOptions(source, target,
+ preserveTimes).withDirectWrite(directWrite || shouldUseDirectWrite())
+ .withAtomicCommit(atomicCommit);
+ runDistCp(builder.build(), true);
+
+ // The source directory itself is in the copy listing, so its times and
+ // those of nested directories must be restored after all concatenations.
+ verifyChunkCopyTree(target, sourceStatuses, preserveTimes, true);
+ }
+
+ @ParameterizedTest(name = "preserveTimes={0}, renameTarget={1}")
+ @CsvSource({"true, false", "false, false", "true, true", "false, true"})
+ public void testChunkCopyOutsideParentScope(boolean preserveTimes,
+ boolean renameTarget)
+ throws Exception {
+ requireChunkCopySupport();
+ Path sourceParent = new Path(remoteDir, "source-parent");
+ Path source = new Path(sourceParent, "file");
+ Path targetParent = new Path(remoteDir, "target-parent");
+ Path target = new Path(targetParent,
+ renameTarget ? "renamed-file" : source.getName());
+ createChunkCopyFile(source, TEST_FILE_LEN * 3 + 1);
+ FileStatus sourceStatus = setChunkCopyTimes(source, 0);
+ FileStatus sourceParentStatus = setChunkCopyTimes(sourceParent, 1);
+ mkdirs(remoteFS, targetParent);
+ FileStatus targetParentStatus = setChunkCopyTimes(targetParent, 2);
+
+ DistCpOptions.Builder builder = chunkCopyOptions(source,
+ renameTarget ? target : targetParent, preserveTimes);
+ if (renameTarget) {
+ builder.withDirectWrite(true);
+ }
+ runDistCp(builder.build(), true);
+
+ assertChunkCopyTimes(sourceStatus, remoteFS.getFileStatus(target),
+ preserveTimes);
+ // Creating/concatenating children may change the destination parent's
+ // mtime naturally. It must not inherit the source parent's timestamps.
+ FileStatus parentAfterCopy = remoteFS.getFileStatus(targetParent);
+ Assertions.assertThat(parentAfterCopy.getModificationTime())
+ .describedAs("mtime of parent outside copy scope")
+ .isNotEqualTo(sourceParentStatus.getModificationTime());
+ assertEquals(targetParentStatus.getAccessTime(),
+ parentAfterCopy.getAccessTime(), "atime of parent outside copy scope");
+ verifyFileContents(remoteFS, target,
+ dataset((int) sourceStatus.getLen(), 'a', 26));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testChunkCopyUpdateRootTimes(boolean updateRoot)
+ throws Exception {
+ requireChunkCopySupport();
+ Path source = new Path(remoteDir, "source");
+ Path target = new Path(remoteDir, "target");
+ Map<String, FileStatus> sourceStatuses = createChunkCopyTree(source);
+ // Force an existing destination file to be replaced by a chunked copy.
+ createFile(remoteFS, new Path(target, "large"), true, new byte[1]);
+ FileStatus targetRootStatus = setChunkCopyTimes(target, 20);
+ DistCpOptions options = chunkCopyOptions(source, target, true)
+ .withSyncFolder(true).withUpdateRoot(updateRoot).build();
+
+ runDistCp(options, true);
+ verifyChunkCopyTree(target, sourceStatuses, true, updateRoot);
+ if (!updateRoot) {
+ assertEquals(targetRootStatus.getAccessTime(),
+ remoteFS.getFileStatus(target).getAccessTime(),
+ "-update must not preserve root times without -updateRoot");
+ }
+
+ // A fully skipped update must still preserve file times without
+ // attempting to concatenate nonexistent chunk files.
+ setChunkCopyTimes(new Path(target, "large"), 30);
+ Job skipped = runDistCp(options, true);
+ assertCounterInRange(skipped, CopyMapper.Counter.COPY, 0, 0);
+ assertCounterInRange(skipped, CopyMapper.Counter.SKIP, 1, -1);
+ assertCounterInRange(skipped, CopyMapper.Counter.BYTESCOPIED, 0, 0);
+ verifyChunkCopyTree(target, sourceStatuses, true, updateRoot);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testChunkCopyPreserveOtherAttributes(boolean preserveTimes)
+ throws Exception {
+ requireChunkCopySupport();
+ skipIfUnsupported(SUPPORTS_UNIX_PERMISSIONS);
+ if (!remoteFS.hasPathCapability(remoteDir, FS_ACLS)
+ || !remoteFS.hasPathCapability(remoteDir, FS_XATTRS)) {
+ skip("ACL and XAttr support required");
+ }
+ Path source = new Path(remoteDir, "source");
+ Path target = new Path(remoteDir, "target");
+ Map<String, FileStatus> sourceStatuses = createChunkCopyTree(source);
+ Path sourceFile = new Path(source, "large");
+ remoteFS.setPermission(sourceFile, new FsPermission((short) 0640));
+ remoteFS.modifyAclEntries(sourceFile, Collections.singletonList(
+ new AclEntry.Builder().setScope(AclEntryScope.ACCESS)
+ .setType(AclEntryType.USER).setName("distcp-reader")
+ .setPermission(FsAction.READ).build()));
+ String xattrName = "user.distcp";
+ byte[] xattrValue = {1, 2, 3};
+ remoteFS.setXAttr(sourceFile, xattrName, xattrValue);
+ sourceStatuses.put("large", setChunkCopyTimes(sourceFile, 0));
+ sourceStatuses.put("", setChunkCopyTimes(source, 11));
+
+ runDistCp(chunkCopyOptions(source, target, preserveTimes)
+ .preserve("rbugpcax").build(), true);
+
+ // File listing entries omit extended ACLs: restoring all attributes from
+ // them after concat would erase ACLs already preserved by the mapper.
+ Path targetFile = new Path(target, "large");
+ assertEquals(remoteFS.getAclStatus(sourceFile).getEntries(),
+ remoteFS.getAclStatus(targetFile).getEntries(), "file ACL entries");
+ assertArrayEquals(xattrValue, remoteFS.getXAttr(targetFile, xattrName));
+ FileStatus expected = sourceStatuses.get("large");
+ FileStatus actual = remoteFS.getFileStatus(targetFile);
+ assertEquals(expected.getPermission(), actual.getPermission());
+ assertEquals(expected.getOwner(), actual.getOwner());
+ assertEquals(expected.getGroup(), actual.getGroup());
+ assertEquals(expected.getReplication(), actual.getReplication());
+ assertEquals(expected.getBlockSize(), actual.getBlockSize());
+ verifyChunkCopyTree(target, sourceStatuses, preserveTimes, true);
+ }
+
+ private void requireChunkCopySupport() throws IOException {
+ skipIfUnsupported(SUPPORTS_CONCAT);
+ skipIfUnsupported(SUPPORTS_SETTIMES);
+ skipIfUnsupported(SUPPORTS_BLOCK_LOCALITY);
+ conf.setInt("mapreduce.local.map.tasks.maximum", 2);
+ }
+
+ private DistCpOptions.Builder chunkCopyOptions(Path source, Path target,
+ boolean preserveTimes) {
+ DistCpOptions.Builder builder = new DistCpOptions.Builder(
+ Collections.singletonList(source), target)
+ .withBlocksPerChunk(1).withDirectWrite(shouldUseDirectWrite());
+ if (preserveTimes) {
+ builder.preserve(FileAttribute.TIMES);
+ }
+ return builder;
+ }
+
+ private Map<String, FileStatus> createChunkCopyTree(Path source)
+ throws IOException {
+ Map<String, FileStatus> statuses = new LinkedHashMap<>();
+ String[] names = {"large", "large2", "nested/large", "block", "small",
+ "empty"};
+ int[] lengths = {TEST_FILE_LEN * 3 + 1, TEST_FILE_LEN * 4,
+ TEST_FILE_LEN * 2, TEST_FILE_LEN, TEST_FILE_LEN / 2, 0};
+ for (int i = 0; i < names.length; i++) {
+ Path file = new Path(source, names[i]);
+ createChunkCopyFile(file, lengths[i]);
+ statuses.put(names[i], setChunkCopyTimes(file, i));
+ }
+ // Set directory times only after all children have been created.
+ statuses.put("nested", setChunkCopyTimes(new Path(source, "nested"), 10));
+ statuses.put("", setChunkCopyTimes(source, 11));
+ return statuses;
+ }
+
+ private void createChunkCopyFile(Path path, int length) throws IOException {
+ writeDataset(remoteFS, path, dataset(length, 'a', 26), length,
+ TEST_FILE_LEN, true);
+ if (length > TEST_FILE_LEN) {
+ FileStatus status = remoteFS.getFileStatus(path);
+ assertTrue(status.getLen() > status.getBlockSize(),
+ "File must exceed its block size to exercise chunked copying");
+ assertTrue(remoteFS.getFileBlockLocations(status, 0, length).length > 1,
+ "File must have multiple blocks to exercise concatenation");
+ }
+ }
+
+ private FileStatus setChunkCopyTimes(Path path, int offset)
+ throws IOException {
+ long mtime = CHUNK_COPY_TIME + offset * MODIFICATION_TIME_OFFSET;
+ remoteFS.setTimes(path, mtime,
+ CHUNK_COPY_ACCESS_TIME + offset * MODIFICATION_TIME_OFFSET);
+ FileStatus status = remoteFS.getFileStatus(path);
+ assertEquals(mtime, status.getModificationTime(), "source fixture mtime");
Review Comment:
The fixture validates only mtime, but later tests assert both mtime and
atime preservation. If `setTimes()` silently ignores atime on a given FS (or
normalizes it), failures later can be harder to diagnose. Consider also
asserting the expected atime when available (e.g., if `status.getAccessTime() >
0`, assert it matches the requested value), or documenting that atime may not
round-trip even when `SUPPORTS_SETTIMES` is true.
##########
hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java:
##########
@@ -654,6 +654,13 @@ private void concatFileChunks(Configuration conf, Path
sourceFile,
DistCpUtils.compareFileLengthsAndChecksums(srcFileStatus.getLen(),
srcfs, sourceFile, null, dstfs,
targetFile, skipCrc, srcFileStatus.getLen());
+ // Concat changes the modification time preserved on the first chunk.
+ // Restore times after validation, which may also update the access time.
+ if (DistCpUtils.unpackAttributes(conf.get(CONF_LABEL_PRESERVE_STATUS))
+ .contains(FileAttribute.TIMES)) {
Review Comment:
`DistCpUtils.unpackAttributes(...)` appears to parse/unpack the
preserved-attributes set on every `concatFileChunks()` call. Since concat may
run once per copied file, this can add avoidable overhead at scale. Consider
unpacking once (e.g., in the committer initialization / caller) and passing a
boolean `preserveTimes`, or caching the unpacked set for reuse.
##########
hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/AbstractContractDistCpTest.java:
##########
@@ -76,6 +89,12 @@ public abstract class AbstractContractDistCpTest
/** Using offset to change modification time in tests. */
private static final long MODIFICATION_TIME_OFFSET = 10000;
+ private static final long CHUNK_COPY_TIME = 1600000000000L;
+
+ // Keep reads and checksum verification from advancing the source atime.
+ private static final long CHUNK_COPY_ACCESS_TIME =
+ (System.currentTimeMillis() / 1000 + 86400) * 1000;
Review Comment:
This makes test behavior time-dependent by setting atime relative to the
wall clock (and into the future). On some filesystems/implementations,
read-atime updates may set atime to 'now' even if that is earlier than the
current atime, which could create non-deterministic failures depending on FS
semantics. A more deterministic approach is to use a fixed constant (like
`CHUNK_COPY_TIME +/- offset`) and, if needed, re-stat the source after the copy
to derive the expected atime (or conditionally assert atime only when it
remains stable on the contract FS).
> Distcp parallel file copy breaks the modification time
> ------------------------------------------------------
>
> Key: HADOOP-17611
> URL: https://issues.apache.org/jira/browse/HADOOP-17611
> Project: Hadoop Common
> Issue Type: Bug
> Reporter: Adam Maroti
> Assignee: Adam Maroti
> Priority: Major
> Labels: pull-request-available
> Time Spent: 2h 50m
> Remaining Estimate: 0h
>
> The commit HADOOP-11794. Enable distcp to copy blocks in parallel.
> (bf3fb585aaf2b179836e139c041fc87920a3c886) broke the modification time of
> large files.
>
> In CopyCommitter.java inside concatFileChunks Filesystem.concat is called
> which changes the modification time therefore the modification times of files
> copeid by distcp will not match the source files. However this only occurs
> for large enough files, which are copied by splitting them up by distcp.
> In concatFileChunks before calling concat extract the modification time and
> apply that to the concatenated result-file after the concat. (probably best
> -after- before the rename()).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
