Fix import ordering, Test in ITestTerasortOnS3A and other minor fix

Author: shameersss1

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -236,6 +236,7 @@
 import static org.apache.hadoop.fs.s3a.auth.delegation.S3ADelegationTokens.hasDelegationTokenBinding;
 import static org.apache.hadoop.fs.s3a.commit.CommitConstants.FS_S3A_COMMITTER_ABORT_PENDING_UPLOADS;
 import static org.apache.hadoop.fs.s3a.commit.CommitConstants.FS_S3A_COMMITTER_STAGING_ABORT_PENDING_UPLOADS;
+import static org.apache.hadoop.fs.s3a.commit.CommitConstants.MAGIC_COMMITTER_PENDING_OBJECT_ETAG_NAME;
 import static org.apache.hadoop.fs.s3a.commit.magic.MagicCommitTrackerUtils.isTrackMagicCommitsInMemoryEnabled;
 import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.submit;
 import static org.apache.hadoop.fs.s3a.impl.CreateFileBuilder.OPTIONS_CREATE_FILE_NO_OVERWRITE;
@@ -3912,14 +3913,14 @@ public FileStatus getFileStatus(final Path f) throws IOException {
       // Some downstream apps might call getFileStatus for a magic path to get the file size.
       // when commit data is stored in memory construct the dummy S3AFileStatus with correct
       // file size fetched from the memory.
-      if (InMemoryMagicCommitTracker.getTaskAttemptIdToBytesWritten().containsKey(path)) {
-        long len = InMemoryMagicCommitTracker.getTaskAttemptIdToBytesWritten().get(path);
+      if (InMemoryMagicCommitTracker.getPathToBytesWritten().containsKey(path)) {
+        long len = InMemoryMagicCommitTracker.getPathToBytesWritten().get(path);
         return new S3AFileStatus(len,
             0L,
             path,
             getDefaultBlockSize(path),
             username,
-            null,
+            MAGIC_COMMITTER_PENDING_OBJECT_ETAG_NAME,
             null);
       }
     }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/CommitConstants.java

@@ -58,6 +58,10 @@ private CommitConstants() {
    */
   public static final String PENDINGSET_SUFFIX = ".pendingset";
 
+  /**
+   * Etag name to be returned on non-committed S3 object: {@value}.
+   */
+  public static final String MAGIC_COMMITTER_PENDING_OBJECT_ETAG_NAME = "pending";
 
   /**
    * Prefix to use for config options: {@value}.
@@ -242,10 +246,15 @@ private CommitConstants() {
    */
   public static final int DEFAULT_COMMITTER_THREADS = 32;
 
-
+  /**
+   * Should Magic committer track all the pending commits in memory?
+   */
   public static final String FS_S3A_COMMITTER_MAGIC_TRACK_COMMITS_IN_MEMORY_ENABLED =
       "fs.s3a.committer.magic.track.commits.in.memory.enabled";
 
+  /**
+   * Default value for {@link #FS_S3A_COMMITTER_MAGIC_TRACK_COMMITS_IN_MEMORY_ENABLED}: {@value}.
+   */
   public static final boolean FS_S3A_COMMITTER_MAGIC_TRACK_COMMITS_IN_MEMORY_ENABLED_DEFAULT =
       false;
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/MagicCommitIntegration.java

@@ -20,14 +20,14 @@
 
 import java.util.List;
 
-import org.apache.hadoop.fs.s3a.commit.magic.InMemoryMagicCommitTracker;
-import org.apache.hadoop.fs.s3a.commit.magic.S3MagicCommitTracker;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.s3a.S3AFileSystem;
 import org.apache.hadoop.fs.s3a.Statistic;
+import org.apache.hadoop.fs.s3a.commit.magic.InMemoryMagicCommitTracker;
+import org.apache.hadoop.fs.s3a.commit.magic.S3MagicCommitTracker;
 import org.apache.hadoop.fs.s3a.impl.AbstractStoreOperation;
 import org.apache.hadoop.fs.s3a.statistics.PutTrackerStatistics;
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/magic/InMemoryMagicCommitTracker.java

@@ -45,15 +45,24 @@
  */
 public class InMemoryMagicCommitTracker extends MagicCommitTracker {
 
-  // stores taskAttemptId to commit data mapping
-  private static Map<String, List<SinglePendingCommit>>
-      taskAttemptIdToMpuMetdadataMap = new ConcurrentHashMap<>();
-
-  // stores the path to its length/size mapping
-  private static Map<Path, Long> taskAttemptIdToBytesWritten = new ConcurrentHashMap<>();
-
-  // stores taskAttemptId to path mapping
-  private static Map<String, List<Path>> taskAttemptIdToPath = new ConcurrentHashMap<>();
+  /**
+   * Map to store taskAttemptId, and it's corresponding list of pending commit data.
+   * The entries in the Map gets removed when a task commits or aborts.
+   */
+  private final static Map<String, List<SinglePendingCommit>>
+      TASK_ATTEMPT_ID_TO_MPU_METDADATA = new ConcurrentHashMap<>();
+
+  /**
+   * Map to store path of the file, and it's corresponding size.
+   * The entries in the Map gets removed when a task commits or aborts.
+   */
+  private final static Map<Path, Long> PATH_TO_BYTES_WRITTEN = new ConcurrentHashMap<>();
+
+  /**
+   * Map to store taskAttemptId, and list of paths to files written by it.
+   * The entries in the Map gets removed when a task commits or aborts.
+   */
+  private final static Map<String, List<Path>> TASK_ATTEMPT_ID_TO_PATH = new ConcurrentHashMap<>();
 
   public InMemoryMagicCommitTracker(Path path,
       String bucket,
@@ -92,16 +101,16 @@ public boolean aboutToComplete(String uploadId,
     String taskAttemptId = extractTaskAttemptIdFromPath(getPath());
 
     // store the commit data with taskAttemptId as the key
-    taskAttemptIdToMpuMetdadataMap.computeIfAbsent(taskAttemptId,
+    TASK_ATTEMPT_ID_TO_MPU_METDADATA.computeIfAbsent(taskAttemptId,
         k -> Collections.synchronizedList(new ArrayList<>())).add(commitData);
 
     // store the byteswritten(length) for the corresponding file
-    taskAttemptIdToBytesWritten.put(getPath(), bytesWritten);
+    PATH_TO_BYTES_WRITTEN.put(getPath(), bytesWritten);
 
     // store the mapping between taskAttemptId and path
     // This information is used for removing entries from
     // the map once the taskAttempt is completed/committed.
-    taskAttemptIdToPath.computeIfAbsent(taskAttemptId,
+    TASK_ATTEMPT_ID_TO_PATH.computeIfAbsent(taskAttemptId,
         k -> Collections.synchronizedList(new ArrayList<>())).add(getPath());
 
     LOG.info("commit metadata for {} parts in {}. size: {} byte(s) "
@@ -113,15 +122,26 @@ public boolean aboutToComplete(String uploadId,
     return false;
   }
 
-  public static Map<String, List<SinglePendingCommit>> getTaskAttemptIdToMpuMetdadataMap() {
-    return taskAttemptIdToMpuMetdadataMap;
+  @Override
+  public String toString() {
+    final StringBuilder sb = new StringBuilder(
+        "InMemoryMagicCommitTracker{");
+    sb.append(", Number of taskAttempts=").append(TASK_ATTEMPT_ID_TO_MPU_METDADATA.size());
+    sb.append(", Number of files=").append(PATH_TO_BYTES_WRITTEN.size());
+    sb.append('}');
+    return sb.toString();
+  }
+
+
+  public static Map<String, List<SinglePendingCommit>> getTaskAttemptIdToMpuMetdadata() {
+    return TASK_ATTEMPT_ID_TO_MPU_METDADATA;
   }
 
-  public static Map<Path, Long> getTaskAttemptIdToBytesWritten() {
-    return taskAttemptIdToBytesWritten;
+  public static Map<Path, Long> getPathToBytesWritten() {
+    return PATH_TO_BYTES_WRITTEN;
   }
 
   public static Map<String, List<Path>> getTaskAttemptIdToPath() {
-    return taskAttemptIdToPath;
+    return TASK_ATTEMPT_ID_TO_PATH;
   }
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/magic/MagicCommitTracker.java

@@ -41,7 +41,7 @@
  * uses any datatype in hadoop-mapreduce.
  */
 @InterfaceAudience.Private
-public class MagicCommitTracker extends PutTracker {
+public abstract class MagicCommitTracker extends PutTracker {
   public static final Logger LOG = LoggerFactory.getLogger(
       MagicCommitTracker.class);
 
@@ -103,22 +103,20 @@ public boolean outputImmediatelyVisible() {
 
   /**
    * Complete operation: generate the final commit data, put it.
-   *
-   * @param uploadId     Upload ID
-   * @param parts        list of parts
+   * @param uploadId Upload ID
+   * @param parts list of parts
    * @param bytesWritten bytes written
    * @param iostatistics nullable IO statistics
    * @return false, indicating that the commit must fail.
-   * @throws IOException              any IO problem.
+   * @throws IOException any IO problem.
    * @throws IllegalArgumentException bad argument
    */
   @Override
-  public boolean aboutToComplete(String uploadId,
+  public abstract boolean aboutToComplete(String uploadId,
       List<CompletedPart> parts,
       long bytesWritten,
-      final IOStatistics iostatistics) throws IOException {
-    return false;
-  }
+      IOStatistics iostatistics)
+      throws IOException;
 
   @Override
   public String toString() {

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/magic/MagicCommitTrackerUtils.java

@@ -18,13 +18,13 @@
 
 package org.apache.hadoop.fs.s3a.commit.magic;
 
+import java.util.List;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.s3a.commit.CommitConstants;
 import org.apache.hadoop.fs.s3a.commit.MagicCommitPaths;
 
-import java.util.List;
-
 import static org.apache.hadoop.util.Preconditions.checkArgument;
 
 /**

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/magic/MagicS3GuardCommitter.java

@@ -18,7 +18,6 @@
 
 package org.apache.hadoop.fs.s3a.commit.magic;
 
-import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -288,7 +287,7 @@ private List<SinglePendingCommit> loadPendingCommitsFromMemory(TaskAttemptContex
     // get all the pending commit metadata associated with the taskAttemptId.
     // This will also remove the entry from the map.
     List<SinglePendingCommit> pendingCommits =
-        InMemoryMagicCommitTracker.getTaskAttemptIdToMpuMetdadataMap().remove(taskAttemptId);
+        InMemoryMagicCommitTracker.getTaskAttemptIdToMpuMetdadata().remove(taskAttemptId);
     // get all the path/files associated with the taskAttemptId.
     // This will also remove the entry from the map.
     List<Path> pathsAssociatedWithTaskAttemptId =
@@ -299,7 +298,7 @@ private List<SinglePendingCommit> loadPendingCommitsFromMemory(TaskAttemptContex
     if (pathsAssociatedWithTaskAttemptId != null) {
       for (Path path : pathsAssociatedWithTaskAttemptId) {
         boolean cleared =
-            InMemoryMagicCommitTracker.getTaskAttemptIdToBytesWritten().remove(path) != null;
+            InMemoryMagicCommitTracker.getPathToBytesWritten().remove(path) != null;
         LOG.debug("Removing path: {} from the memory isSuccess: {}", path, cleared);
       }
     } else {

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/terasort/ITestTerasortOnS3A.java

@@ -44,6 +44,7 @@
 import org.apache.hadoop.examples.terasort.TeraValidate;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.s3a.commit.AbstractYarnClusterITest;
+import org.apache.hadoop.fs.s3a.commit.CommitConstants;
 import org.apache.hadoop.fs.s3a.commit.magic.MagicS3GuardCommitter;
 import org.apache.hadoop.fs.s3a.commit.staging.DirectoryStagingCommitter;
 import org.apache.hadoop.mapred.JobConf;
@@ -97,6 +98,9 @@ public class ITestTerasortOnS3A extends AbstractYarnClusterITest {
   /** Name of the committer for this run. */
   private final String committerName;
 
+  /** Should Magic committer track pending commits in-memory. */
+  private final boolean trackCommitsInMemory;
+
   /** Base path for all the terasort input and output paths. */
   private Path terasortPath;
 
@@ -117,12 +121,14 @@ public class ITestTerasortOnS3A extends AbstractYarnClusterITest {
   @Parameterized.Parameters(name = "{0}")
   public static Collection<Object[]> params() {
     return Arrays.asList(new Object[][]{
-        {DirectoryStagingCommitter.NAME},
-        {MagicS3GuardCommitter.NAME}});
+        {DirectoryStagingCommitter.NAME, false},
+        {MagicS3GuardCommitter.NAME, false},
+        {MagicS3GuardCommitter.NAME, true}});
   }
 
-  public ITestTerasortOnS3A(final String committerName) {
+  public ITestTerasortOnS3A(final String committerName, final boolean trackCommitsInMemory) {
     this.committerName = committerName;
+    this.trackCommitsInMemory = trackCommitsInMemory;
   }
 
   @Override
@@ -152,6 +158,9 @@ protected void applyCustomConfigOptions(JobConf conf) {
     conf.setBoolean(
         TeraSortConfigKeys.USE_SIMPLE_PARTITIONER.key(),
         false);
+    conf.setBoolean(
+        CommitConstants.FS_S3A_COMMITTER_MAGIC_TRACK_COMMITS_IN_MEMORY_ENABLED,
+        trackCommitsInMemory);
   }
 
   private int getExpectedPartitionCount() {
@@ -173,7 +182,7 @@ protected int getRowCount() {
    */
   private void prepareToTerasort() {
     // small sample size for faster runs
-    terasortPath = new Path("/terasort-" + committerName)
+    terasortPath = new Path("/terasort-" + committerName + "-" + trackCommitsInMemory)
         .makeQualified(getFileSystem());
     sortInput = new Path(terasortPath, "sortin");
     sortOutput = new Path(terasortPath, "sortout");