HADOOP-19354. S3AInputStream to be created by factory under S3AStore

Ability to create custom streams (type = custom), which
reads class from "fs.s3a.input.stream.custom.factory".
This is mainly for testing, especially CNFE and similar.

Unit test TestStreamFactories for this.

ObjectInputStreams save and export stream type to assist
these tests too, as it enables assertions on the generated
stream type.

Simplified that logic related to the old prefetch enabled flag

If fs.s3a.prefetch.enabled is true, the prefetch stream is returned,
the stream.type option is not used at all. Simpler logic, simpler
docs, fewer support calls.

Parameters supplied to ObjectInputStreamFactory.bind converted
to a parameter object. Allows for more parameters to be added later
if ever required.

ObjectInputStreamFactory returns more requirements to
the store/fs. For this reason
  StreamThreadOptions threadRequirements();
is renamed
  StreamFactoryRequirements factoryRequirements()

VectorIO context changes
* Returned in factoryRequirements()
* exiting configuration reading code moved into
  StreamIntegration.populateVectoredIOContext()
* Streams which don't have custom vector IO, e.g. prefetching
  can return a minimum seek range of 0.
  This disables range merging on the default PositionedReadable
  implementation, so ensures that they will only get asked for
  data which will be read...leaving prefetch/cache code
  to know exactly what is needed.

Other
 * Draft docs.
 * Stream capability declares stream type
   & is exported through FS too.
   (todo: test, document, add to bucket-info)
 * ConfigurationHelper.resolveEnum() supercedes
   Configuration.getEnum() with
     - case independence
     - fallback is a supplier<Enum> rather than a simple
       value.

Change-Id: I2e59300af48042df8173de61d0b3d6139a0ae7fe

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/ConfigurationHelper.java

@@ -22,6 +22,7 @@
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -123,4 +124,39 @@ public static <E extends Enum<E>> Map<String, E> mapEnumNamesToValues(
     return mapping;
   }
 
+  /**
+   * Look up an enum from the configuration option and map it to
+   * a value in the supplied enum class.
+   * If no value is supplied or there is no match for the supplied value,
+   * the fallback function is invoked, passing in the trimmed and possibly
+   * empty string of the value.
+   * Extends {link {@link Configuration#getEnum(String, Enum)}}
+   * by adding case independence and a lambda expression for fallback,
+   * rather than a default value.
+   * @param conf configuration
+   * @param name property name
+   * @param enumClass classname to resolve
+   * @param fallback fallback supplier
+   * @return an enum value
+   * @param <E> enumeration type.
+   * @throws IllegalArgumentException If mapping is illegal for the type provided
+   */
+    public static <E extends Enum<E>> E resolveEnum(
+        Configuration conf,
+        String name,
+        Class<E> enumClass,
+        Function<String, E> fallback) {
+
+      final String val = conf.getTrimmed(name, "");
+
+      // build a map of lower case string to enum values.
+      final Map<String, E> mapping = mapEnumNamesToValues("", enumClass);
+      final E mapped = mapping.get(val.toLowerCase(Locale.ROOT));
+      if (mapped != null) {
+        return mapped;
+      } else {
+        // fallback handles it
+        return fallback.apply(val);
+      }
+    }
 }

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

@@ -21,7 +21,6 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.Options;
-import org.apache.hadoop.fs.s3a.impl.streams.InputStreamType;
 import org.apache.hadoop.fs.s3a.impl.streams.StreamIntegration;
 import org.apache.hadoop.security.ssl.DelegatingSSLSocketFactory;
 
@@ -1594,30 +1593,48 @@ private Constants() {
       StreamIntegration.CLASSIC;
 
   /**
-   * The prefetching input stream: "prefetch".
+   * The prefetching input stream: {@value}.
    */
   public static final String INPUT_STREAM_TYPE_PREFETCH = StreamIntegration.PREFETCH;
 
   /**
-   * The analytics input stream: "analytics".
+   * The analytics input stream: {@value}.
    */
   public static final String INPUT_STREAM_TYPE_ANALYTICS =
       StreamIntegration.ANALYTICS;
 
   /**
-   * The default input stream.
-   * Currently {@link #INPUT_STREAM_TYPE_CLASSIC}
+   * Request the default input stream,
+   * whatever it is for this release: {@value}.
    */
-  public static final String INPUT_STREAM_TYPE_DEFAULT = InputStreamType.DEFAULT_STREAM_TYPE.getName();
+  public static final String INPUT_STREAM_TYPE_DEFAULT = StreamIntegration.DEFAULT;
+
+  /**
+   * The custom input stream type: {@value}".
+   * If set, the classname is loaded from
+   * {@link #INPUT_STREAM_CUSTOM_FACTORY}.
+   * <p>
+   * This option is primarily for testing as it can
+   * be used to generated failures.
+   */
+  public static final String INPUT_STREAM_TYPE_CUSTOM =
+      StreamIntegration.CUSTOM;
+
+  /**
+   * Classname of the factory to instantiate for custom streams: {@value}.
+   */
+  public static final String INPUT_STREAM_CUSTOM_FACTORY = "fs.s3a.input.stream.custom.factory";
 
   /**
    * Controls whether the prefetching input stream is enabled.
    */
+  @Deprecated
   public static final String PREFETCH_ENABLED_KEY = "fs.s3a.prefetch.enabled";
 
   /**
    * Default option as to whether the prefetching input stream is enabled.
    */
+  @Deprecated
   public static final boolean  PREFETCH_ENABLED_DEFAULT = false;
 
   // If the default values are used, each file opened for reading will consume

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

@@ -148,7 +148,7 @@
 import org.apache.hadoop.fs.s3a.impl.CSEUtils;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectReadParameters;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStreamCallbacks;
-import org.apache.hadoop.fs.s3a.impl.streams.StreamThreadOptions;
+import org.apache.hadoop.fs.s3a.impl.streams.StreamFactoryRequirements;
 import org.apache.hadoop.fs.s3a.tools.MarkerToolOperations;
 import org.apache.hadoop.fs.s3a.tools.MarkerToolOperationsImpl;
 import org.apache.hadoop.fs.statistics.DurationTracker;
@@ -369,12 +369,6 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
   /** Vectored IO context. */
   private VectoredIOContext vectoredIOContext;
 
-  /**
-   * Maximum number of active range read operation a single
-   * input stream can have.
-   */
-  private int vectoredActiveRangeReads;
-
   private long readAhead;
   private ChangeDetectionPolicy changeDetectionPolicy;
   private final AtomicBoolean closed = new AtomicBoolean(false);
@@ -755,9 +749,6 @@ public void initialize(URI name, Configuration originalConf)
           longBytesOption(conf, ASYNC_DRAIN_THRESHOLD,
                         DEFAULT_ASYNC_DRAIN_THRESHOLD, 0),
           inputPolicy);
-      vectoredActiveRangeReads = intOption(conf,
-              AWS_S3_VECTOR_ACTIVE_RANGE_READS, DEFAULT_AWS_S3_VECTOR_ACTIVE_RANGE_READS, 1);
-      vectoredIOContext = populateVectoredIOContext(conf);
       scheme = (this.uri != null && this.uri.getScheme() != null) ? this.uri.getScheme() : FS_S3A;
       optimizedCopyFromLocal = conf.getBoolean(OPTIMIZED_COPY_FROM_LOCAL,
           OPTIMIZED_COPY_FROM_LOCAL_DEFAULT);
@@ -772,6 +763,11 @@ public void initialize(URI name, Configuration originalConf)
       // this is to aid mocking.
       s3Client = getStore().getOrCreateS3Client();
 
+      final StreamFactoryRequirements factoryRequirements =
+          getStore().factoryRequirements();
+      // get the vector IO context from the factory.
+      vectoredIOContext = factoryRequirements.vectoredIOContext();
+
       // thread pool init requires store to be created
       initThreadPools();
 
@@ -843,23 +839,6 @@ protected S3AStore createS3AStore(final ClientManager clientManager,
     return st;
   }
 
-  /**
-   * Populates the configurations related to vectored IO operation
-   * in the context which has to passed down to input streams.
-   * @param conf configuration object.
-   * @return VectoredIOContext.
-   */
-  private VectoredIOContext populateVectoredIOContext(Configuration conf) {
-    final int minSeekVectored = (int) longBytesOption(conf, AWS_S3_VECTOR_READS_MIN_SEEK_SIZE,
-            DEFAULT_AWS_S3_VECTOR_READS_MIN_SEEK_SIZE, 0);
-    final int maxReadSizeVectored = (int) longBytesOption(conf, AWS_S3_VECTOR_READS_MAX_MERGED_READ_SIZE,
-            DEFAULT_AWS_S3_VECTOR_READS_MAX_MERGED_READ_SIZE, 0);
-    return new VectoredIOContext()
-            .setMinSeekForVectoredReads(minSeekVectored)
-            .setMaxReadSizeForVectoredReads(maxReadSizeVectored)
-            .build();
-  }
-
   /**
    * Test bucket existence in S3.
    * When the value of {@link Constants#S3A_BUCKET_PROBE} is set to 0,
@@ -950,9 +929,9 @@ private void initThreadPools() {
             TimeUnit.SECONDS,
             Duration.ZERO).getSeconds();
 
-    final StreamThreadOptions threadRequirements =
-        getStore().threadRequirements();
-    int numPrefetchThreads = threadRequirements.sharedThreads();
+    final StreamFactoryRequirements factoryRequirements =
+        getStore().factoryRequirements();
+    int numPrefetchThreads = factoryRequirements.sharedThreads();
 
     int activeTasksForBoundedThreadPool = maxThreads;
     int waitingTasksForBoundedThreadPool = maxThreads + totalTasks + numPrefetchThreads;
@@ -970,7 +949,7 @@ private void initThreadPools() {
     unboundedThreadPool.allowCoreThreadTimeOut(true);
     executorCapacity = intOption(conf,
         EXECUTOR_CAPACITY, DEFAULT_EXECUTOR_CAPACITY, 1);
-    if (threadRequirements.createFuturePool()) {
+    if (factoryRequirements.createFuturePool()) {
       // create a future pool.
       final S3AInputStreamStatistics s3AInputStreamStatistics =
           statisticsContext.newInputStreamStatistics();
@@ -1860,13 +1839,13 @@ private FSDataInputStream executeOpen(
     LOG.debug("Opening '{}'", readContext);
 
     // what does the stream need
-    final StreamThreadOptions requirements =
-        getStore().threadRequirements();
+    final StreamFactoryRequirements requirements =
+        getStore().factoryRequirements();
 
     // calculate the permit count.
     final int permitCount = requirements.streamThreads() +
         (requirements.vectorSupported()
-            ? vectoredActiveRangeReads
+            ? requirements.vectoredIOContext().getVectoredActiveRangeReads()
             : 0);
     // create an executor which is a subset of the
     // bounded thread pool.

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

@@ -146,9 +146,6 @@ public class S3AInputStream extends ObjectInputStream implements CanSetReadahead
 
   private long readahead = Constants.DEFAULT_READAHEAD_RANGE;
 
-  /** Vectored IO context. */
-  private final VectoredIOContext vectoredIOContext;
-
   /**
    * This is the actual position within the object, used by
    * lazy seek to decide whether to seek on the next read or not.
@@ -195,7 +192,6 @@ public S3AInputStream(ObjectReadParameters parameters) {
         getObjectAttributes());
     setReadahead(context.getReadahead());
     this.asyncDrainThreshold = context.getAsyncDrainThreshold();
-    this.vectoredIOContext = this.getContext().getVectoredIOContext();
   }
 
   /**
@@ -764,6 +760,7 @@ public String toString() {
     synchronized (this) {
       final StringBuilder sb = new StringBuilder(
           "S3AInputStream{");
+      sb.append(super.toString()).append(" ");
       sb.append(getUri());
       sb.append(" wrappedStream=")
           .append(isObjectStreamOpen() ? "open" : "closed");
@@ -776,7 +773,7 @@ public String toString() {
       sb.append(" remainingInCurrentRequest=")
           .append(remainingInCurrentRequest());
       sb.append(" ").append(changeTracker);
-      sb.append(" ").append(vectoredIOContext);
+      sb.append(" ").append(getVectoredIOContext());
       sb.append('\n').append(s);
       sb.append('}');
       return sb.toString();
@@ -826,22 +823,6 @@ public void readFully(long position, byte[] buffer, int offset, int length)
     }
   }
 
-  /**
-   * {@inheritDoc}.
-   */
-  @Override
-  public int minSeekForVectorReads() {
-    return vectoredIOContext.getMinSeekForVectorReads();
-  }
-
-  /**
-   * {@inheritDoc}.
-   */
-  @Override
-  public int maxReadSizeForVectorReads() {
-    return vectoredIOContext.getMaxReadSizeForVectorReads();
-  }
-
   /**
    * {@inheritDoc}
    * Vectored read implementation for S3AInputStream.

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

@@ -40,6 +40,12 @@ public class VectoredIOContext {
    */
   private int maxReadSizeForVectorReads;
 
+  /**
+   * Maximum number of active range read operation a single
+   * input stream can have.
+   */
+  private int vectoredActiveRangeReads;
+
   /**
    * Default no arg constructor.
    */
@@ -68,11 +74,22 @@ public int getMaxReadSizeForVectorReads() {
     return maxReadSizeForVectorReads;
   }
 
+  public int getVectoredActiveRangeReads() {
+    return vectoredActiveRangeReads;
+  }
+
+  public VectoredIOContext setVectoredActiveRangeReads(
+      final int vectoredActiveRangeReads) {
+    this.vectoredActiveRangeReads = vectoredActiveRangeReads;
+    return this;
+  }
+
   @Override
   public String toString() {
     return "VectoredIOContext{" +
             "minSeekForVectorReads=" + minSeekForVectorReads +
             ", maxReadSizeForVectorReads=" + maxReadSizeForVectorReads +
+            ", vectoredActiveRangeReads=" + vectoredActiveRangeReads +
             '}';
   }
 }

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

@@ -73,10 +73,12 @@
 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.streams.FactoryBindingParameters;
+import org.apache.hadoop.fs.s3a.impl.streams.InputStreamType;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStream;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStreamFactory;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectReadParameters;
-import org.apache.hadoop.fs.s3a.impl.streams.StreamThreadOptions;
+import org.apache.hadoop.fs.s3a.impl.streams.StreamFactoryRequirements;
 import org.apache.hadoop.fs.s3a.statistics.S3AStatisticsContext;
 import org.apache.hadoop.fs.statistics.DurationTracker;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
@@ -111,7 +113,7 @@
 import static org.apache.hadoop.fs.s3a.Statistic.STORE_IO_THROTTLE_RATE;
 import static org.apache.hadoop.fs.s3a.impl.ErrorTranslation.isObjectNotFound;
 import static org.apache.hadoop.fs.s3a.impl.InternalConstants.DELETE_CONSIDERED_IDEMPOTENT;
-import static org.apache.hadoop.fs.s3a.impl.streams.StreamIntegration.createStreamFactory;
+import static org.apache.hadoop.fs.s3a.impl.streams.StreamIntegration.factoryFromConfig;
 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;
@@ -233,7 +235,7 @@ protected void serviceInit(final Configuration conf) throws Exception {
 
     // create and register the stream factory, which will
     // then follow the service lifecycle
-    objectInputStreamFactory = createStreamFactory(conf);
+    objectInputStreamFactory = factoryFromConfig(conf);
     addService(objectInputStreamFactory);
 
     // init all child services, including the stream factory
@@ -269,7 +271,6 @@ public boolean hasPathCapability(final Path path, final String capability) {
     }
   }
 
-
   /**
    * Return the capabilities of input streams created
    * through the store.
@@ -955,7 +956,7 @@ private void finishStreamFactoryInit() {
         "Client Manager is in wrong state: %s", clientManager.getServiceState());
 
     // finish initialization and pass down callbacks to self
-    objectInputStreamFactory.bind(new FactoryCallbacks());
+    objectInputStreamFactory.bind(new FactoryBindingParameters(new FactoryCallbacks()));
   }
 
   @Override /* ObjectInputStreamFactory */
@@ -966,21 +967,25 @@ public ObjectInputStream readObject(ObjectReadParameters parameters)
   }
 
   @Override /* ObjectInputStreamFactory */
-  public StreamThreadOptions threadRequirements() {
-    return objectInputStreamFactory.threadRequirements();
+  public StreamFactoryRequirements factoryRequirements() {
+    return objectInputStreamFactory.factoryRequirements();
   }
 
   /**
    * This operation is not implemented, as
    * is this class which invokes it on the actual factory.
-   * @param callbacks factory callbacks.
-   * @throws UnsupportedOperationException always
+   * @param factoryBindingParameters@throws UnsupportedOperationException always
    */
   @Override /* ObjectInputStreamFactory */
-  public void bind(final StreamFactoryCallbacks callbacks) {
+  public void bind(final FactoryBindingParameters factoryBindingParameters) {
     throw new UnsupportedOperationException("Not supported");
   }
 
+  @Override /* ObjectInputStreamFactory */
+  public InputStreamType streamType() {
+    return objectInputStreamFactory.streamType();
+  }
+
   /**
    * Callbacks from {@link ObjectInputStreamFactory} instances.
    */