Allow passing of factory params

Author: rajdchak

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

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a.impl.streams;
+
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+
+/**
+ * Factory parameters class to hold various parameters needed for stream creation.
+ */
+public class FactoryParams {
+    private final S3AsyncClient s3AsyncClient;
+
+    private FactoryParams(Builder builder) {
+        this.s3AsyncClient = builder.s3AsyncClient;
+    }
+
+    public S3AsyncClient getS3AsyncClient() {
+        return s3AsyncClient;
+    }
+
+    public static class Builder {
+        private S3AsyncClient s3AsyncClient;
+
+        public Builder withS3AsyncClient(S3AsyncClient s3AsyncClient) {
+            this.s3AsyncClient = s3AsyncClient;
+            return this;
+        }
+
+        public FactoryParams build() {
+            return new FactoryParams(this);
+        }
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+}

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

@@ -23,7 +23,6 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.s3a.prefetch.PrefetchingInputStreamFactory;
-import software.amazon.awssdk.services.s3.S3AsyncClient;
 
 /**
  * Enum of input stream types.
@@ -45,14 +44,14 @@ public enum InputStreamType {
   /**
    * The analytics input stream.
    */
-  Analytics("analytics", (c, s3AsyncClient) -> new S3ASeekableInputStreamFactory(s3AsyncClient));
+  Analytics("analytics", (c, factoryParams) -> new S3ASeekableInputStreamFactory(factoryParams.getS3AsyncClient()));
 
   /**
    * Name.
    */
   private final String name;
 
-  private final BiFunction<Configuration, S3AsyncClient, ObjectInputStreamFactory> factory;
+  private final BiFunction<Configuration, FactoryParams, ObjectInputStreamFactory> factory;
   /**
    * String name.
    * @return the name
@@ -65,7 +64,7 @@ public String getName() {
     this(name, (c, s) -> factory.apply(c));
   }
 
-  InputStreamType(String name, BiFunction<Configuration, S3AsyncClient, ObjectInputStreamFactory> factory) {
+  InputStreamType(String name, BiFunction<Configuration, FactoryParams, ObjectInputStreamFactory> factory) {
     this.name = name;
     this.factory = factory;
   }
@@ -74,7 +73,7 @@ public String getName() {
    * Factory constructor.
    * @return the factory associated with this stream type.
    */
-  public BiFunction<Configuration, S3AsyncClient, ObjectInputStreamFactory> factory() {
+  public BiFunction<Configuration, FactoryParams, ObjectInputStreamFactory> factory() {
     return factory;
   }
 

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

@@ -63,13 +63,17 @@ public static ObjectInputStreamFactory createStreamFactory(final Configuration c
 
   /**
    * Create the s3 seekable input stream factory.
+   * @param conf configuration
    * @param s3AsyncClient s3 async client
    * @return a stream factory.
    */
   public static ObjectInputStreamFactory createStreamFactory(final Configuration conf, final S3AsyncClient s3AsyncClient) {
+    FactoryParams factoryParams = FactoryParams.builder()
+            .withS3AsyncClient(s3AsyncClient)
+            .build();
     InputStreamType defaultStream = InputStreamType.DEFAULT_STREAM_TYPE;
     return conf.getEnum(INPUT_STREAM_TYPE, defaultStream)
             .factory()
-            .apply(conf, s3AsyncClient);  }
+            .apply(conf, factoryParams);  }
 
 }