HADOOP-18028. High performance S3A input stream

Initial merge tuning

* rename test classes, have AbstractHadoopTestBase as the base
* package info files for new packages
* import ordering
* move to intercept() for assertions; ExceptionAsserts is invoking it
  and can be removed in future.

Tests still fail; will document in JIRA

Change-Id: Ib80edd72c30af8dc996acb62aa0e31bf61eebc7a

Author: steveloughran

hadoop-tools/hadoop-aws/pom.xml

@@ -350,10 +350,6 @@
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-surefire-plugin</artifactId>
         <configuration>
-          <includes>
-            <include>**/*Test.java</include>
-            <include>**/Test*.java</include>
-          </includes>
           <forkedProcessTimeoutInSeconds>3600</forkedProcessTimeoutInSeconds>
           <systemPropertyVariables>
             <test.default.timeout>${test.integration.timeout}</test.default.timeout>

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/common/BlockOperations.java

@@ -19,9 +19,6 @@
 
 package org.apache.hadoop.fs.common;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.DoubleSummaryStatistics;
@@ -31,6 +28,9 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * Block level operations performed on a file.
  * This class is meant to be used by {@code BlockManager}.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/common/BoundedResourcePool.java

@@ -24,8 +24,6 @@
 import java.util.Set;
 import java.util.concurrent.ArrayBlockingQueue;
 
-import static org.apache.hadoop.util.Preconditions.checkNotNull;
-
 /**
  * Manages a fixed pool of resources.
  *
@@ -81,7 +79,7 @@ public T tryAcquire() {
    */
   @Override
   public void release(T item) {
-    checkNotNull(item, "item");
+    Validate.checkNotNull(item, "item");
 
     synchronized (this.createdItems) {
       if (!this.createdItems.contains(item)) {

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/common/BufferData.java

@@ -19,16 +19,16 @@
 
 package org.apache.hadoop.fs.common;
 
-import com.twitter.util.Future;
-import com.twitter.util.Awaitable.CanAwait;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.zip.CRC32;
 
+import com.twitter.util.Awaitable.CanAwait;
+import com.twitter.util.Future;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * Holds the state of a ByteBuffer that is in use by {@code CachingBlockManager}.
  *

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/common/BufferPool.java

@@ -19,10 +19,6 @@
 
 package org.apache.hadoop.fs.common;
 
-import com.twitter.util.Future;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.Closeable;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -32,9 +28,9 @@
 import java.util.Map;
 import java.util.concurrent.CancellationException;
 
-import static org.apache.hadoop.util.Preconditions.checkArgument;
-import static org.apache.hadoop.util.Preconditions.checkNotNull;
-import static org.apache.hadoop.util.Preconditions.checkState;
+import com.twitter.util.Future;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Manages a fixed pool of {@code ByteBuffer} instances.
@@ -155,7 +151,7 @@ private synchronized BufferData acquireHelper(int blockNumber, boolean canBlock)
     data = new BufferData(blockNumber, buffer.duplicate());
 
     synchronized (this.allocated) {
-      checkState(this.find(blockNumber) == null, "buffer data already exists");
+      Validate.checkState(this.find(blockNumber) == null, "buffer data already exists");
 
       this.allocated.put(data, buffer);
     }
@@ -212,10 +208,10 @@ private int distance(BufferData data, int blockNumber) {
    * @throws IllegalArgumentException if data cannot be released due to its state.
    */
   public synchronized void release(BufferData data) {
-    checkNotNull(data, "data");
+    Validate.checkNotNull(data, "data");
 
     synchronized (data) {
-      checkArgument(
+      Validate.checkArgument(
           this.canRelease(data),
           String.format("Unable to release buffer: %s", data));
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/common/CachingBlockManager.java

@@ -19,18 +19,18 @@
 
 package org.apache.hadoop.fs.common;
 
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
 import com.twitter.util.Await;
 import com.twitter.util.ExceptionalFunction0;
 import com.twitter.util.Future;
 import com.twitter.util.FuturePool;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-
 /**
  * Provides read access to the underlying file one block at a time.
  * Improve read performance by prefetching and locall caching blocks.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/common/FilePosition.java

@@ -21,8 +21,6 @@
 
 import java.nio.ByteBuffer;
 
-import static org.apache.hadoop.util.Preconditions.checkState;
-
 /**
  * Provides functionality related to tracking the position within a file.
  *
@@ -278,7 +276,7 @@ public String toString() {
 
   private void throwIfInvalidBuffer() {
     if (!this.isValid()) {
-      checkState(buffer != null, "'buffer' must not be null");
+      Validate.checkState(buffer != null, "'buffer' must not be null");
     }
   }
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/common/SingleFilePerBlockCache.java

@@ -19,9 +19,6 @@
 
 package org.apache.hadoop.fs.common;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.File;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -42,6 +39,9 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * Provides functionality necessary for caching blocks of data read from FileSystem.
  * Each cache block is stored on the local disk as a separate file.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/common/Validate.java

@@ -23,8 +23,6 @@
 import java.nio.file.Path;
 import java.util.Collection;
 
-import org.apache.hadoop.util.Preconditions;
-
 /**
  * A superset of Validate class in Apache commons lang3.
  *
@@ -42,7 +40,7 @@ private Validate() {}
    * @param argName the name of the argument being validated.
    */
   public static void checkNotNull(Object obj, String argName) {
-    Preconditions.checkNotNull(obj != null, "'%s' must not be null.", argName);
+    checkArgument(obj != null, "'%s' must not be null.", argName);
   }
 
   /**
@@ -52,7 +50,7 @@ public static void checkNotNull(Object obj, String argName) {
    * @param argName the name of the argument being validated.
    */
   public static void checkPositiveInteger(long value, String argName) {
-    Preconditions.checkArgument(value > 0, "'%s' must be a positive integer.", argName);
+    checkArgument(value > 0, "'%s' must be a positive integer.", argName);
   }
 
   /**
@@ -62,7 +60,7 @@ public static void checkPositiveInteger(long value, String argName) {
    * @param argName the name of the argument being validated.
    */
   public static void checkNotNegative(long value, String argName) {
-    Preconditions.checkArgument(value >= 0, "'%s' must not be negative.", argName);
+    checkArgument(value >= 0, "'%s' must not be negative.", argName);
   }
 
   /*
@@ -72,7 +70,7 @@ public static void checkNotNegative(long value, String argName) {
    * @param argName the name of the argument being validated.
    */
   public static void checkRequired(boolean isPresent, String argName) {
-    Preconditions.checkArgument(isPresent, "'%s' is required.", argName);
+    checkArgument(isPresent, "'%s' is required.", argName);
   }
 
   /**
@@ -82,7 +80,7 @@ public static void checkRequired(boolean isPresent, String argName) {
    * @param argName the name of the argument being validated.
    */
   public static void checkValid(boolean isValid, String argName) {
-    Preconditions.checkArgument(isValid, "'%s' is invalid.", argName);
+    checkArgument(isValid, "'%s' is invalid.", argName);
   }
 
   /**
@@ -387,16 +385,19 @@ public static void checkPathExistsAsDir(Path path, String argName) {
    */
   public static void checkPathExistsAsFile(Path path, String argName) {
     checkPathExists(path, argName);
-    Preconditions.checkArgument(Files.isRegularFile(path),
-        "Path %s (%s) must point to a file.", argName, path);
+    checkArgument(Files.isRegularFile(path), "Path %s (%s) must point to a file.", argName, path);
   }
 
   public static void checkArgument(boolean expression, String format, Object... args) {
-    Preconditions.checkArgument(expression, format, args);
+    org.apache.commons.lang3.Validate.isTrue(expression, format, args);
+  }
+
+  public static void checkState(boolean expression, String format, Object... args) {
+    org.apache.commons.lang3.Validate.validState(expression, format, args);
   }
 
   private static void checkNotEmpty(int arraySize, String argName) {
-    Preconditions.checkArgument(
+    Validate.checkArgument(
         arraySize > 0,
         "'%s' must have at least one element.",
         argName);

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

@@ -24,8 +24,7 @@
 
 import org.apache.hadoop.fs.common.BlockData;
 import org.apache.hadoop.fs.common.BlockManager;
-
-import static org.apache.hadoop.util.Preconditions.checkNotNull;
+import org.apache.hadoop.fs.common.Validate;
 
 /**
  * Provides read access to S3 file one block at a time.
@@ -52,7 +51,9 @@ public class S3BlockManager extends BlockManager {
   public S3BlockManager(S3Reader reader, BlockData blockData) {
     super(blockData);
 
-    this.reader = checkNotNull(reader, "reader");
+    Validate.checkNotNull(reader, "reader");
+
+    this.reader = reader;
   }
 
   /**