HADOOP-19688. S3A: ITestS3ACommitterMRJob failing on JUnit5 (#7968)

Fixes the ITestS3ACommitterMRJob test suite to work on JUnit 5, through changes
in local test directory creation and use.

MapReduce FileInputFormat provides some more information when raising NPEs
from null block location lists.

Contributed by Steve Loughran

Author: steveloughran

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.java

@@ -49,6 +49,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static java.util.Objects.requireNonNull;
 import static org.apache.hadoop.fs.FileUtil.maybeIgnoreMissingDirectory;
 
 /**
@@ -455,10 +456,17 @@ public List<InputSplit> getSplits(JobContext job) throws IOException {
       if (length != 0) {
         BlockLocation[] blkLocations;
         if (file instanceof LocatedFileStatus) {
-          blkLocations = ((LocatedFileStatus) file).getBlockLocations();
+          blkLocations = requireNonNull(
+              ((LocatedFileStatus) file).getBlockLocations(),
+              () -> String.format("No block locations in LocatedFileStatus %s; length=%d",
+                  file, length));
+
         } else {
           FileSystem fs = path.getFileSystem(job.getConfiguration());
-          blkLocations = fs.getFileBlockLocations(file, 0, length);
+          blkLocations = requireNonNull(fs.getFileBlockLocations(file, 0, length), () ->
+              String.format("No block locations returned from getFileBlockLocations(%s, 0, %d)"
+                      + "; status=%s",
+                  path, length, file));
         }
         if (isSplitable(job, path)) {
           long blockSize = file.getBlockSize();

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

@@ -95,15 +95,15 @@ public S3AFileStatus toS3AFileStatus() {
 
   @Override
   public String toString() {
-    final StringBuilder sb = new StringBuilder(
-        super.toString());
+    final StringBuilder sb = new StringBuilder("S3ALocatedFileStatus{");
+    sb.append(super.toString());
     sb.append("[eTag='").
         append(eTag != null ? eTag : "")
         .append('\'');
     sb.append(", versionId='")
         .append(versionId != null ? versionId: "")
         .append('\'');
-    sb.append(']');
+    sb.append('}');
     return sb.toString();
   }
 }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/integration/ITestS3ACommitterMRJob.java

@@ -22,6 +22,7 @@
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.net.URI;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -74,6 +75,7 @@
 
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.disableFilesystemCaching;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.lsR;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
 import static org.apache.hadoop.fs.s3a.S3AUtils.applyLocatedFiles;
 import static org.apache.hadoop.fs.s3a.commit.CommitConstants.FS_S3A_COMMITTER_STAGING_TMP_PATH;
 import static org.apache.hadoop.fs.s3a.commit.CommitConstants.MAGIC_PATH_PREFIX;
@@ -82,6 +84,7 @@
 import static org.apache.hadoop.fs.s3a.commit.staging.Paths.getMultipartUploadCommitsDirectory;
 import static org.apache.hadoop.fs.s3a.commit.staging.StagingCommitterConstants.STAGING_UPLOADS;
 import static org.apache.hadoop.mapred.JobConf.MAPRED_TASK_ENV;
+import static org.apache.hadoop.mapreduce.lib.input.FileInputFormat.LIST_STATUS_NUM_THREADS;
 
 /**
  * Test an MR Job with all the different committers.
@@ -105,24 +108,24 @@
  *   <li>
  *     The test suites are declared to be executed in ascending order, so
  *     that for a specific binding, the order is
- *     {@link #test_000(CommitterTestBinding)},
- *     {@link #test_100(CommitterTestBinding)}
- *     {@link #test_200_execute(CommitterTestBinding, java.nio.file.Path)} and finally
- *     {@link #test_500(CommitterTestBinding)}.
+ *     {@link #test_000()},
+ *     {@link #test_100()}
+ *     {@link #test_200_execute()} and finally
+ *     {@link #test_500()}.
  *   </li>
  *   <li>
- *     {@link #test_000(CommitterTestBinding)} calls
+ *     {@link #test_000()} calls
  *     {@link CommitterTestBinding#validate()} to
  *     as to validate the state of the committer. This is primarily to
  *     verify that the binding setup mechanism is working.
  *   </li>
  *   <li>
- *     {@link #test_100(CommitterTestBinding)} is relayed to
+ *     {@link #test_100()} is relayed to
  *     {@link CommitterTestBinding#test_100()},
  *     for any preflight tests.
  *   </li>
  *   <li>
- *     The {@link #test_200_execute(CommitterTestBinding, java.nio.file.Path)}
+ *     The {@link #test_200_execute()}
  *     test runs the MR job for that
  *     particular binding with standard reporting and verification of the
  *     outcome.
@@ -165,6 +168,9 @@ public static Collection<Object[]> params() {
    */
   private final CommitterTestBinding committerTestBinding;
 
+  @TempDir
+  private java.nio.file.Path localFilesDir;
+
   /**
    * Parameterized constructor.
    * @param committerTestBinding binding for the test.
@@ -186,6 +192,9 @@ public void setup() throws Exception {
   protected Configuration createConfiguration() {
     Configuration conf = super.createConfiguration();
     disableFilesystemCaching(conf);
+    removeBaseAndBucketOverrides(conf,
+        LIST_STATUS_NUM_THREADS);
+    conf.setInt(LIST_STATUS_NUM_THREADS, 16);
     return conf;
   }
 
@@ -208,9 +217,9 @@ public void test_100() throws Throwable {
   }
 
   @Test
-  public void test_200_execute(
-      @TempDir java.nio.file.Path localFilesDir) throws Exception {
+  public void test_200_execute() throws Exception {
     describe("Run an MR with committer %s", committerName());
+    LOG.info("Local Temp directory is {}", localFilesDir);
 
     S3AFileSystem fs = getFileSystem();
     // final dest is in S3A
@@ -253,8 +262,10 @@ public void test_200_execute(
     jobConf.set(FS_S3A_COMMITTER_UUID, commitUUID);
 
     mrJob.setInputFormatClass(TextInputFormat.class);
-    FileInputFormat.addInputPath(mrJob,
-        new Path(localFilesDir.getRoot().toUri()));
+
+    final URI inputPath = localFilesDir.toUri();
+    LOG.info("Job input path {}", inputPath);
+    FileInputFormat.addInputPath(mrJob, new Path(inputPath));
 
     mrJob.setMapperClass(MapClass.class);
     mrJob.setNumReduceTasks(0);