initial commit

Change-Id: If42bdd0b227c4da07c62a410a998e6d8c35581f6

Author: ahmarsuhail

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

@@ -1560,6 +1560,11 @@ private Constants() {
    */
   public static final String AWS_AUTH_CLASS_PREFIX = "com.amazonaws.auth";
 
+
+  public static final String INPUT_STREAM_TYPE = "fs.s3a.input.stream.type";
+
+  public static final String INPUT_STREAM_TYPE_DEFAULT = "classic";
+
   /**
    * Controls whether the prefetching input stream is enabled.
    */

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

@@ -0,0 +1,29 @@
+package org.apache.hadoop.fs.s3a;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public enum InputStreamType {
+  CLASSIC("classic"),
+  PREFETCH("prefetch"),
+  ANALYTICS("analytics");
+
+  private final String name;
+
+  private static final Logger LOG = LoggerFactory.getLogger(InputStreamType.class);
+
+  InputStreamType(String name) {
+    this.name = name;
+  }
+
+  public static InputStreamType fromString(String inputStreamType) {
+    for (InputStreamType value : values()) {
+      if (value.name.equalsIgnoreCase(inputStreamType)) {
+        return value;
+      }
+    }
+    LOG.warn("Unknown input stream type {}, using default classic stream.", inputStreamType);
+
+    return CLASSIC;
+  }
+}

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

@@ -126,6 +126,7 @@
 import org.apache.hadoop.fs.s3a.impl.ContextAccessors;
 import org.apache.hadoop.fs.s3a.impl.CopyFromLocalOperation;
 import org.apache.hadoop.fs.s3a.impl.CreateFileBuilder;
+import org.apache.hadoop.fs.s3a.impl.InputStreamCallbacksImpl;
 import org.apache.hadoop.fs.s3a.impl.S3AFileSystemOperations;
 import org.apache.hadoop.fs.s3a.impl.CSEV1CompatibleS3AFileSystemOperations;
 import org.apache.hadoop.fs.s3a.impl.CSEMaterials;
@@ -1884,6 +1885,7 @@ private FSDataInputStream executeOpen(
         auditSpan);
     fileInformation.applyOptions(readContext);
     LOG.debug("Opening '{}'", readContext);
+    // QUESTION: why are we creating a new executor on each open?
     final SemaphoredDelegatingExecutor pool = new SemaphoredDelegatingExecutor(
         boundedThreadPool,
         vectoredActiveRangeReads,
@@ -1901,21 +1903,7 @@ private FSDataInputStream executeOpen(
     // TODO: move into S3AStore and export the factory API through
     // the store, which will add some of the features (callbacks, stats)
     // before invoking the real factory
-    ObjectInputStreamFactory factory = null;
-    try {
-      // Choose factory.
-      if (prefetchEnabled) {
-        factory = new PrefetchingInputStreamFactory();
-      } else {
-        factory = new ClassicObjectInputStreamFactory();
-      }
-      factory.init(getConf());
-      factory.start();
-      return new FSDataInputStream(factory.readObject(parameters));
-    } finally {
-      IOUtils.cleanupWithLogger(LOG, factory);
-    }
-
+    return new FSDataInputStream(getStore().readObject(parameters));
   }
 
   /**
@@ -1924,68 +1912,9 @@ private FSDataInputStream executeOpen(
    */
   private ObjectInputStreamCallbacks createInputStreamCallbacks(
       final AuditSpan auditSpan) {
-    return new InputStreamCallbacksImpl(auditSpan);
+    return new InputStreamCallbacksImpl(auditSpan, getStore(), fsHandler, unboundedThreadPool);
   }
 
-  /**
-   * Operations needed by ObjectInputStreams to read data.
-   */
-  private final class InputStreamCallbacksImpl implements
-      ObjectInputStreamCallbacks {
-
-    /**
-     * Audit span to activate before each call.
-     */
-    private final AuditSpan auditSpan;
-
-    /**
-     * Create.
-     * @param auditSpan Audit span to activate before each call.
-     */
-    private InputStreamCallbacksImpl(final AuditSpan auditSpan) {
-      this.auditSpan = requireNonNull(auditSpan);
-    }
-
-    /**
-     * Closes the audit span.
-     */
-    @Override
-    public void close()  {
-      auditSpan.close();
-    }
-
-    @Override
-    public GetObjectRequest.Builder newGetRequestBuilder(final String key) {
-      // active the audit span used for the operation
-      try (AuditSpan span = auditSpan.activate()) {
-        return getRequestFactory().newGetObjectRequestBuilder(key);
-      }
-    }
-
-    @Override
-    public ResponseInputStream<GetObjectResponse> getObject(GetObjectRequest request) throws
-        IOException {
-      // active the audit span used for the operation
-      try (AuditSpan span = auditSpan.activate()) {
-        return fsHandler.getObject(getStore(), request, getRequestFactory());
-      }
-    }
-
-    @Override
-    public <T> CompletableFuture<T> submit(final CallableRaisingIOE<T> operation) {
-      CompletableFuture<T> result = new CompletableFuture<>();
-      unboundedThreadPool.submit(() ->
-          LambdaUtils.eval(result, () -> {
-            LOG.debug("Starting submitted operation in {}", auditSpan.getSpanId());
-            try (AuditSpan span = auditSpan.activate()) {
-              return operation.apply();
-            } finally {
-              LOG.debug("Completed submitted operation in {}", auditSpan.getSpanId());
-            }
-          }));
-      return result;
-    }
-  }
 
   /**
    * Callbacks for WriteOperationHelper.

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

@@ -46,13 +46,16 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.s3a.api.RequestFactory;
 import org.apache.hadoop.fs.s3a.impl.ChangeTracker;
 import org.apache.hadoop.fs.s3a.impl.ClientManager;
 import org.apache.hadoop.fs.s3a.impl.MultiObjectDeleteException;
 import org.apache.hadoop.fs.s3a.impl.S3AFileSystemOperations;
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
+import org.apache.hadoop.fs.s3a.impl.model.ObjectInputStreamFactory;
+import org.apache.hadoop.fs.s3a.impl.model.ObjectReadParameters;
 import org.apache.hadoop.fs.s3a.statistics.S3AStatisticsContext;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
 import org.apache.hadoop.fs.statistics.IOStatisticsSource;
@@ -73,7 +76,7 @@
  */
 @InterfaceAudience.LimitedPrivate("Extensions")
 @InterfaceStability.Unstable
-public interface S3AStore extends Service, IOStatisticsSource, ClientManager {
+public interface S3AStore extends Service, IOStatisticsSource, ClientManager, ObjectInputStreamFactory {
 
   /**
    * Acquire write capacity for operations.

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

@@ -0,0 +1,87 @@
+package org.apache.hadoop.fs.s3a.impl;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.core.ResponseInputStream;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+
+import org.apache.hadoop.fs.s3a.S3AStore;
+import org.apache.hadoop.fs.s3a.impl.model.ObjectInputStreamCallbacks;
+import org.apache.hadoop.fs.store.audit.AuditSpan;
+import org.apache.hadoop.util.LambdaUtils;
+import org.apache.hadoop.util.functional.CallableRaisingIOE;
+
+import static java.util.Objects.requireNonNull;
+
+public class InputStreamCallbacksImpl implements ObjectInputStreamCallbacks {
+
+  /**
+   * Audit span to activate before each call.
+   */
+  private final AuditSpan auditSpan;
+
+  private final S3AStore store;
+
+  private final S3AFileSystemOperations fsHandler;
+
+  private static final Logger LOG = LoggerFactory.getLogger(InputStreamCallbacksImpl.class);
+
+  private final ThreadPoolExecutor unboundedThreadPool;
+
+  /**
+   * Create.
+   * @param auditSpan Audit span to activate before each call.
+   */
+  public InputStreamCallbacksImpl(final AuditSpan auditSpan, final S3AStore store,
+      S3AFileSystemOperations fsHandler, ThreadPoolExecutor unboundedThreadPool) {
+    this.auditSpan = requireNonNull(auditSpan);
+    this.store = requireNonNull(store);
+    this.fsHandler = requireNonNull(fsHandler);
+    this.unboundedThreadPool = requireNonNull(unboundedThreadPool);
+  }
+
+  /**
+   * Closes the audit span.
+   */
+  @Override
+  public void close()  {
+    auditSpan.close();
+  }
+
+  @Override
+  public GetObjectRequest.Builder newGetRequestBuilder(final String key) {
+    // active the audit span used for the operation
+    try (AuditSpan span = auditSpan.activate()) {
+      return store.getRequestFactory().newGetObjectRequestBuilder(key);
+    }
+  }
+
+  @Override
+  public ResponseInputStream<GetObjectResponse> getObject(GetObjectRequest request) throws
+      IOException {
+    // active the audit span used for the operation
+    try (AuditSpan span = auditSpan.activate()) {
+      return fsHandler.getObject(store, request, store.getRequestFactory());
+    }
+  }
+
+  @Override
+  public <T> CompletableFuture<T> submit(final CallableRaisingIOE<T> operation) {
+    CompletableFuture<T> result = new CompletableFuture<>();
+    unboundedThreadPool.submit(() ->
+        LambdaUtils.eval(result, () -> {
+          LOG.debug("Starting submitted operation in {}", auditSpan.getSpanId());
+          try (AuditSpan span = auditSpan.activate()) {
+            return operation.apply();
+          } finally {
+            LOG.debug("Completed submitted operation in {}", auditSpan.getSpanId());
+          }
+        }));
+    return result;
+  }
+}

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

@@ -58,9 +58,11 @@
 import software.amazon.awssdk.transfer.s3.model.UploadFileRequest;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.InputStreamType;
 import org.apache.hadoop.fs.s3a.Invoker;
 import org.apache.hadoop.fs.s3a.ProgressableProgressListener;
 import org.apache.hadoop.fs.s3a.Retries;
@@ -72,6 +74,10 @@
 import org.apache.hadoop.fs.s3a.UploadInfo;
 import org.apache.hadoop.fs.s3a.api.RequestFactory;
 import org.apache.hadoop.fs.s3a.audit.AuditSpanS3A;
+import org.apache.hadoop.fs.s3a.impl.model.ObjectInputStream;
+import org.apache.hadoop.fs.s3a.impl.model.ObjectInputStreamFactory;
+import org.apache.hadoop.fs.s3a.impl.model.ObjectReadParameters;
+import org.apache.hadoop.fs.s3a.prefetch.PrefetchingInputStreamFactory;
 import org.apache.hadoop.fs.s3a.statistics.S3AStatisticsContext;
 import org.apache.hadoop.fs.statistics.DurationTracker;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
@@ -85,6 +91,8 @@
 import static java.util.Objects.requireNonNull;
 import static org.apache.hadoop.fs.s3a.Constants.BUFFER_DIR;
 import static org.apache.hadoop.fs.s3a.Constants.HADOOP_TMP_DIR;
+import static org.apache.hadoop.fs.s3a.Constants.INPUT_STREAM_TYPE;
+import static org.apache.hadoop.fs.s3a.Constants.INPUT_STREAM_TYPE_DEFAULT;
 import static org.apache.hadoop.fs.s3a.S3AUtils.extractException;
 import static org.apache.hadoop.fs.s3a.S3AUtils.getPutRequestLength;
 import static org.apache.hadoop.fs.s3a.S3AUtils.isThrottleException;
@@ -108,6 +116,7 @@
 import static org.apache.hadoop.fs.statistics.StoreStatisticNames.ACTION_HTTP_GET_REQUEST;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDurationOfOperation;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDurationOfSupplier;
+import static org.apache.hadoop.io.IOUtils.cleanupWithLogger;
 import static org.apache.hadoop.util.Preconditions.checkArgument;
 
 /**
@@ -116,7 +125,7 @@
  * to move.
  */
 public class S3AStoreImpl extends CompositeService
-    implements S3AStore {
+    implements S3AStore, ObjectInputStreamFactory {
 
   private static final Logger LOG = LoggerFactory.getLogger(S3AStoreImpl.class);
 
@@ -177,6 +186,8 @@ public class S3AStoreImpl extends CompositeService
    */
   private LocalDirAllocator directoryAllocator;
 
+  private ObjectInputStreamFactory objectInputStreamFactory;
+
   /** Constructor to create S3A store. */
   S3AStoreImpl(StoreContextFactory storeContextFactory,
       ClientManager clientManager,
@@ -209,12 +220,14 @@ public class S3AStoreImpl extends CompositeService
   protected void serviceStart() throws Exception {
     super.serviceStart();
     initLocalDirAllocator();
+    initInputStreamFactory();
   }
 
   @Override
   protected void serviceStop() throws Exception {
     super.serviceStop();
     clientManager.close();
+    cleanupWithLogger(LOG, objectInputStreamFactory);
   }
 
   /**
@@ -226,6 +239,20 @@ private void initLocalDirAllocator() {
     directoryAllocator = new LocalDirAllocator(bufferDir);
   }
 
+  private void initInputStreamFactory() {
+    InputStreamType inputStreamType = InputStreamType.fromString(getConfig().get(INPUT_STREAM_TYPE, INPUT_STREAM_TYPE_DEFAULT));
+    switch (inputStreamType) {
+    case PREFETCH:
+      this.objectInputStreamFactory = new PrefetchingInputStreamFactory();
+      break;
+    default:
+      this.objectInputStreamFactory = new ClassicObjectInputStreamFactory();
+    }
+
+    this.objectInputStreamFactory.init(getConfig());
+    this.objectInputStreamFactory.start();
+  }
+
   /** Acquire write capacity for rate limiting {@inheritDoc}. */
   @Override
   public Duration acquireWriteCapacity(final int capacity) {
@@ -871,4 +898,15 @@ public File createTemporaryFileForWriting(String pathStr,
     return File.createTempFile(prefix, null, dir);
   }
 
+  @Override
+  public ObjectInputStream readObject(ObjectReadParameters parameters)
+      throws IOException {
+       if (objectInputStreamFactory != null) {
+          return objectInputStreamFactory.readObject(parameters);
+        } else {
+          // TODO: Find the right exception to throw if factory has not yet been initialised, or closed.
+          throw new IOException("Factory not initialized!");
+        }
+  }
+
 }