HADOOP-18596. review comments

Author: mehakmeet

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpConstants.java

@@ -143,8 +143,13 @@ private DistCpConstants() {
 
   public static final String CONF_LABEL_USE_ITERATOR = "distcp.use.iterator";
 
-  /** Distcp -update to use modification time of source and target file to
-   * check while skipping.
+  /**
+   * Enabling distcp -update to use modification time of source and target
+   * file to check while copying same file with same size but different content.
+   *
+   * The check would verify if the target file is perceived as older than the
+   * source then it indicates that the source has been recently updated and it
+   * is a newer version than what was synced, so we should not skip the copy.
    */
   public static final String CONF_LABEL_UPDATE_MOD_TIME =
       "distcp.update.modification.time";

hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/AbstractContractDistCpTest.java

@@ -50,6 +50,7 @@
 import org.apache.hadoop.tools.mapred.CopyMapper;
 import org.apache.hadoop.tools.util.DistCpTestUtils;
 import org.apache.hadoop.util.functional.RemoteIterators;
+import org.apache.http.annotation.Contract;
 
 import org.assertj.core.api.Assertions;
 import org.junit.Before;
@@ -359,6 +360,29 @@ private Job distCpUpdate(final Path srcDir, final Path destDir)
             .withOverwrite(false)));
   }
 
+  /**
+   * Run distcp -update srcDir destDir.
+   * @param srcDir local source directory
+   * @param destDir remote destination directory.
+   * @return the completed job
+   * @throws Exception any failure.
+   */
+  private Job distCpUpdateWithFs(final Path srcDir, final Path destDir,
+      FileSystem sourceFs, FileSystem targetFs)
+      throws Exception {
+    describe("\nDistcp -update from " + srcDir + " to " + destDir);
+    lsR("Source Fs to update", sourceFs, srcDir);
+    lsR("Target Fs before update", targetFs, destDir);
+    return runDistCp(buildWithStandardOptions(
+        new DistCpOptions.Builder(
+            Collections.singletonList(srcDir), destDir)
+            .withDeleteMissing(true)
+            .withSyncFolder(true)
+            .withSkipCRC(true)
+            .withDirectWrite(shouldUseDirectWrite())
+            .withOverwrite(false)));
+  }
+
   /**
    * Update the source directories as various tests expect,
    * including adding a new file.
@@ -868,35 +892,41 @@ public void testDistCpUpdateCheckFileSkip() throws Exception {
 
     Path source = new Path(remoteDir, "file");
     Path dest = new Path(localDir, "file");
+
+    Path source_0byte = new Path(remoteDir, "file_0byte");
+    Path dest_0byte = new Path(localDir, "file_0byte");
     dest = localFS.makeQualified(dest);
+    dest_0byte = localFS.makeQualified(dest_0byte);
 
     // Creating a source file with certain dataset.
     byte[] sourceBlock = dataset(10, 'a', 'z');
 
     // Write the dataset and as well create the target path.
-    try (FSDataOutputStream out = remoteFS.create(source)) {
-      out.write(sourceBlock);
-      localFS.create(dest);
-    }
+    ContractTestUtils.createFile(localFS, dest, true, sourceBlock);
+    ContractTestUtils
+        .writeDataset(remoteFS, source, sourceBlock, sourceBlock.length,
+            1024, true);
 
-    verifyPathExists(remoteFS, "", source);
-    verifyPathExists(localFS, "", dest);
-    DistCpTestUtils
-        .assertRunDistCp(DistCpConstants.SUCCESS, remoteDir.toString(),
-            localDir.toString(), "-delete -update" + getDefaultCLIOptions(),
-            conf);
+    // Create 0 byte source and target files.
+    ContractTestUtils.createFile(remoteFS, source_0byte, true, new byte[0]);
+    ContractTestUtils.createFile(localFS, dest_0byte, true, new byte[0]);
+
+    // Execute the distcp -update job.
+    Job job = distCpUpdateWithFs(remoteDir, localDir, remoteFS, localFS);
 
     // First distcp -update would normally copy the source to dest.
     verifyFileContents(localFS, dest, sourceBlock);
+    // Verify 1 file was skipped in the distcp -update(They 0 byte files).
+    // Verify 1 file was copied in the distcp -update(The new source file).
+    verifySkipAndCopyCounter(job, 1, 1);
 
     // Remove the source file and replace with a file with same name and size
     // but different content.
     remoteFS.delete(source, false);
     Path updatedSource = new Path(remoteDir, "file");
     byte[] updatedSourceBlock = dataset(10, 'b', 'z');
-    try (FSDataOutputStream out = remoteFS.create(updatedSource)) {
-      out.write(updatedSourceBlock);
-    }
+    ContractTestUtils.writeDataset(remoteFS, updatedSource,
+        updatedSourceBlock, updatedSourceBlock.length, 1024, true);
 
     // For testing purposes we would take the modification time of the
     // updated Source file and add an offset or subtract the offset and set
@@ -913,32 +943,65 @@ public void testDistCpUpdateCheckFileSkip() throws Exception {
     long newTargetModTimeNew = modTimeSourceUpd + MODIFICATION_TIME_OFFSET;
     localFS.setTimes(dest, newTargetModTimeNew, -1);
 
-    DistCpTestUtils
-        .assertRunDistCp(DistCpConstants.SUCCESS, remoteDir.toString(),
-            localDir.toString(), "-delete -update" + getDefaultCLIOptions(),
-            conf);
+    // Execute the distcp -update job.
+    Job updatedSourceJobOldSrc =
+        distCpUpdateWithFs(remoteDir, localDir, remoteFS,
+            localFS);
 
     // File contents should remain same since the mod time for target is
     // newer than the updatedSource which indicates that the sync happened
     // more recently and there is no update.
     verifyFileContents(localFS, dest, sourceBlock);
+    // Skipped both 0 byte file and sourceFile(since mod time of target is
+    // older than the source it is perceived that source is of older version
+    // and we can skip it's copy).
+    verifySkipAndCopyCounter(updatedSourceJobOldSrc, 2, 0);
 
     // Subtract by an offset which would ensure enough gap for the test to
     // not fail due to race conditions.
     long newTargetModTimeOld =
         Math.min(modTimeSourceUpd - MODIFICATION_TIME_OFFSET, 0);
     localFS.setTimes(dest, newTargetModTimeOld, -1);
 
-    DistCpTestUtils
-        .assertRunDistCp(DistCpConstants.SUCCESS, remoteDir.toString(),
-            localDir.toString(), "-delete -update" + getDefaultCLIOptions(),
-            conf);
+    // Execute the distcp -update job.
+    Job updatedSourceJobNewSrc = distCpUpdateWithFs(remoteDir, localDir,
+        remoteFS,
+        localFS);
 
-    Assertions.assertThat(RemoteIterators.toList(localFS.listFiles(dest, true)))
-        .hasSize(1);
+    // Verifying the target directory have both 0 byte file and the content
+    // file.
+    Assertions
+        .assertThat(RemoteIterators.toList(localFS.listFiles(localDir, true)))
+        .hasSize(2);
     // Now the copy should take place and the file contents should change
     // since the mod time for target is older than the source file indicating
     // that there was an update to the source after the last sync took place.
     verifyFileContents(localFS, dest, updatedSourceBlock);
+    // Verifying we skipped the 0 byte file and copied the updated source
+    // file (since the modification time of the new source is older than the
+    // target now).
+    verifySkipAndCopyCounter(updatedSourceJobNewSrc, 1, 1);
+  }
+
+  /**
+   * Method to check the skipped and copied counters of a distcp job.
+   *
+   * @param job               job to check.
+   * @param skipExpectedValue expected skip counter value.
+   * @param copyExpectedValue expected copy counter value.
+   * @throws IOException throw in case of failures.
+   */
+  private void verifySkipAndCopyCounter(Job job,
+      int skipExpectedValue, int copyExpectedValue) throws IOException {
+    // get the skip and copy counters from the job.
+    long skipActualValue = job.getCounters()
+        .findCounter(CopyMapper.Counter.SKIP).getValue();
+    long copyActualValue = job.getCounters()
+        .findCounter(CopyMapper.Counter.COPY).getValue();
+    // Verify if the actual values equals the expected ones.
+    assertEquals("Mismatch in COPY counter value", copyExpectedValue,
+        copyActualValue);
+    assertEquals("Mismatch in SKIP counter value", skipExpectedValue,
+        skipActualValue);
   }
 }