Merge pull request apache#14 from passaro/HADOOP-18073-v2/select

Upgrade select to SDK v2

Author: passaro

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

@@ -31,6 +31,7 @@
 import software.amazon.awssdk.core.retry.RetryPolicy;
 import software.amazon.awssdk.http.apache.ApacheHttpClient;
 import software.amazon.awssdk.http.apache.ProxyConfiguration;
+import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
 import software.amazon.awssdk.thirdparty.org.apache.http.client.utils.URIBuilder;
 
 import org.apache.hadoop.conf.Configuration;
@@ -111,6 +112,33 @@ public static ApacheHttpClient.Builder createHttpClientBuilder(Configuration con
     return httpClientBuilder;
   }
 
+  /**
+   * Configures the async http client.
+   *
+   * @param conf The Hadoop configuration
+   * @return Http client builder
+   */
+  public static NettyNioAsyncHttpClient.Builder createAsyncHttpClientBuilder(Configuration conf) {
+    NettyNioAsyncHttpClient.Builder httpClientBuilder =
+        NettyNioAsyncHttpClient.builder();
+
+    httpClientBuilder.maxConcurrency(S3AUtils.intOption(conf, MAXIMUM_CONNECTIONS,
+        DEFAULT_MAXIMUM_CONNECTIONS, 1));
+
+    int connectionEstablishTimeout =
+        S3AUtils.intOption(conf, ESTABLISH_TIMEOUT, DEFAULT_ESTABLISH_TIMEOUT, 0);
+    int socketTimeout = S3AUtils.intOption(conf, SOCKET_TIMEOUT, DEFAULT_SOCKET_TIMEOUT, 0);
+
+    httpClientBuilder.connectionTimeout(Duration.ofSeconds(connectionEstablishTimeout));
+    httpClientBuilder.readTimeout(Duration.ofSeconds(socketTimeout));
+    httpClientBuilder.writeTimeout(Duration.ofSeconds(socketTimeout));
+
+    // TODO: Need to set ssl socket factory, as done in
+    //  NetworkBinding.bindSSLChannelMode(conf, awsConf);
+
+    return httpClientBuilder;
+  }
+
   /**
    * Configures the retry policy.
    *
@@ -132,10 +160,10 @@ public static RetryPolicy.Builder createRetryPolicyBuilder(Configuration conf) {
    *
    * @param conf The Hadoop configuration
    * @param bucket Optional bucket to use to look up per-bucket proxy secrets
-   * @return Proxy configuration builder
+   * @return Proxy configuration
    * @throws IOException on any IO problem
    */
-  public static ProxyConfiguration.Builder createProxyConfigurationBuilder(Configuration conf,
+  public static ProxyConfiguration createProxyConfiguration(Configuration conf,
       String bucket) throws IOException {
 
     ProxyConfiguration.Builder proxyConfigBuilder = ProxyConfiguration.builder();
@@ -181,7 +209,72 @@ public static ProxyConfiguration.Builder createProxyConfigurationBuilder(Configu
       throw new IllegalArgumentException(msg);
     }
 
-    return proxyConfigBuilder;
+    return proxyConfigBuilder.build();
+  }
+
+  /**
+   * Configures the proxy for the async http client.
+   *
+   * @param conf The Hadoop configuration
+   * @param bucket Optional bucket to use to look up per-bucket proxy secrets
+   * @return Proxy configuration
+   * @throws IOException on any IO problem
+   */
+  public static software.amazon.awssdk.http.nio.netty.ProxyConfiguration
+      createAsyncProxyConfiguration(Configuration conf,
+      String bucket) throws IOException {
+
+    software.amazon.awssdk.http.nio.netty.ProxyConfiguration.Builder proxyConfigBuilder =
+        software.amazon.awssdk.http.nio.netty.ProxyConfiguration.builder();
+
+    String proxyHost = conf.getTrimmed(PROXY_HOST, "");
+    int proxyPort = conf.getInt(PROXY_PORT, -1);
+
+    if (!proxyHost.isEmpty()) {
+      if (proxyPort >= 0) {
+        proxyConfigBuilder.host(proxyHost);
+        proxyConfigBuilder.port(proxyPort);
+      } else {
+        if (conf.getBoolean(SECURE_CONNECTIONS, DEFAULT_SECURE_CONNECTIONS)) {
+          LOG.warn("Proxy host set without port. Using HTTPS default 443");
+          proxyConfigBuilder.host(proxyHost);
+          proxyConfigBuilder.port(443);
+        } else {
+          LOG.warn("Proxy host set without port. Using HTTP default 80");
+          proxyConfigBuilder.host(proxyHost);
+          proxyConfigBuilder.port(80);
+        }
+      }
+      final String proxyUsername = S3AUtils.lookupPassword(bucket, conf, PROXY_USERNAME,
+          null, null);
+      final String proxyPassword = S3AUtils.lookupPassword(bucket, conf, PROXY_PASSWORD,
+          null, null);
+      if ((proxyUsername == null) != (proxyPassword == null)) {
+        String msg = "Proxy error: " + PROXY_USERNAME + " or " +
+            PROXY_PASSWORD + " set without the other.";
+        LOG.error(msg);
+        throw new IllegalArgumentException(msg);
+      }
+      proxyConfigBuilder.username(proxyUsername);
+      proxyConfigBuilder.password(proxyPassword);
+      // TODO: check NTLM support
+      // proxyConfigBuilder.ntlmDomain(conf.getTrimmed(PROXY_DOMAIN));
+      // proxyConfigBuilder.ntlmWorkstation(conf.getTrimmed(PROXY_WORKSTATION));
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Using proxy server {}:{} as user {} with password {} on "
+                + "domain {} as workstation {}", proxyHost, proxyPort, proxyUsername, proxyPassword,
+            PROXY_DOMAIN, PROXY_WORKSTATION);
+      }
+    } else if (proxyPort >= 0) {
+      String msg =
+          "Proxy error: " + PROXY_PORT + " set without " + PROXY_HOST;
+      LOG.error(msg);
+      throw new IllegalArgumentException(msg);
+    } else {
+      return null;
+    }
+
+    return proxyConfigBuilder.build();
   }
 
   /***

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

@@ -55,10 +55,11 @@
 import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
 import software.amazon.awssdk.core.retry.RetryPolicy;
 import software.amazon.awssdk.http.apache.ApacheHttpClient;
-import software.amazon.awssdk.http.apache.ProxyConfiguration;
+import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
 import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3BaseClientBuilder;
 import software.amazon.awssdk.services.s3.S3Client;
-import software.amazon.awssdk.services.s3.S3ClientBuilder;
 import software.amazon.awssdk.services.s3.S3Configuration;
 import software.amazon.awssdk.services.s3.model.HeadBucketRequest;
 import software.amazon.awssdk.services.s3.model.HeadBucketResponse;
@@ -184,34 +185,85 @@ public AmazonS3 createS3Client(
     }
   }
 
-  /**
-   * Creates a new {@link S3Client}.
-   *
-   * @param uri S3A file system URI
-   * @param parameters parameter object
-   * @return S3 client
-   * @throws IOException on any IO problem
-   */
   @Override
   public S3Client createS3ClientV2(
       final URI uri,
       final S3ClientCreationParameters parameters) throws IOException {
 
     Configuration conf = getConf();
     bucket = uri.getHost();
+    ApacheHttpClient.Builder httpClientBuilder = AWSClientConfig
+        .createHttpClientBuilder(conf)
+        .proxyConfiguration(AWSClientConfig.createProxyConfiguration(conf, bucket));
+    return configureClientBuilder(S3Client.builder(), parameters, conf, bucket)
+        .httpClientBuilder(httpClientBuilder)
+        .build();
+  }
 
-    final ClientOverrideConfiguration.Builder clientOverrideConfigBuilder =
-        AWSClientConfig.createClientConfigBuilder(conf);
+  @Override
+  public S3AsyncClient createS3AsyncClient(
+      final URI uri,
+      final S3ClientCreationParameters parameters) throws IOException {
 
-    final ApacheHttpClient.Builder httpClientBuilder =
-        AWSClientConfig.createHttpClientBuilder(conf);
+    Configuration conf = getConf();
+    bucket = uri.getHost();
+    NettyNioAsyncHttpClient.Builder httpClientBuilder = AWSClientConfig
+        .createAsyncHttpClientBuilder(conf)
+        .proxyConfiguration(AWSClientConfig.createAsyncProxyConfiguration(conf, bucket));
+    return configureClientBuilder(S3AsyncClient.builder(), parameters, conf, bucket)
+        .httpClientBuilder(httpClientBuilder)
+        .build();
+  }
 
-    final RetryPolicy.Builder retryPolicyBuilder = AWSClientConfig.createRetryPolicyBuilder(conf);
+  /**
+   * Configure a sync or async S3 client builder.
+   * This method handles all shared configuration.
+   * @param builder S3 client builder
+   * @param parameters parameter object
+   * @param conf configuration object
+   * @param bucket bucket name
+   * @return the builder object
+   * @param <BuilderT> S3 client builder type
+   * @param <ClientT> S3 client type
+   */
+  private static <BuilderT extends S3BaseClientBuilder<BuilderT, ClientT>, ClientT>
+  BuilderT configureClientBuilder(
+      BuilderT builder,
+      S3ClientCreationParameters parameters,
+      Configuration conf,
+      String bucket) {
 
-    final ProxyConfiguration.Builder proxyConfigBuilder =
-        AWSClientConfig.createProxyConfigurationBuilder(conf, bucket);
+    URI endpoint = getS3Endpoint(parameters.getEndpoint(), conf);
+    Region region = getS3Region(conf.getTrimmed(AWS_REGION), bucket,
+        parameters.getCredentialSet());
+    LOG.debug("Using endpoint {}; and region {}", endpoint, region);
 
-    S3ClientBuilder s3ClientBuilder = S3Client.builder();
+    // TODO: Some configuration done in configureBasicParams is not done yet.
+    S3Configuration serviceConfiguration = S3Configuration.builder()
+        .pathStyleAccessEnabled(parameters.isPathStyleAccess())
+        .build();
+
+    return builder
+        .overrideConfiguration(createClientOverrideConfiguration(parameters, conf))
+        .credentialsProvider(
+            // use adapter classes so V1 credential providers continue to work. This will
+            // be moved to AWSCredentialProviderList.add() when that class is updated.
+            V1V2AwsCredentialProviderAdapter.adapt(parameters.getCredentialSet()))
+        .endpointOverride(endpoint)
+        .region(region)
+        .serviceConfiguration(serviceConfiguration);
+  }
+
+  /**
+   * Create an override configuration for an S3 client.
+   * @param parameters parameter object
+   * @param conf configuration object
+   * @return the override configuration
+   */
+  private static ClientOverrideConfiguration createClientOverrideConfiguration(
+      S3ClientCreationParameters parameters, Configuration conf) {
+    final ClientOverrideConfiguration.Builder clientOverrideConfigBuilder =
+        AWSClientConfig.createClientConfigBuilder(conf);
 
     // add any headers
     parameters.getHeaders().forEach((h, v) -> clientOverrideConfigBuilder.putHeader(h, v));
@@ -232,40 +284,12 @@ public S3Client createS3ClientV2(
       }
     }
 
+    final RetryPolicy.Builder retryPolicyBuilder = AWSClientConfig.createRetryPolicyBuilder(conf);
     clientOverrideConfigBuilder.retryPolicy(retryPolicyBuilder.build());
-    httpClientBuilder.proxyConfiguration(proxyConfigBuilder.build());
-
-    s3ClientBuilder.httpClientBuilder(httpClientBuilder)
-        .overrideConfiguration(clientOverrideConfigBuilder.build());
-
-    // use adapter classes so V1 credential providers continue to work. This will be moved to
-    // AWSCredentialProviderList.add() when that class is updated.
-    s3ClientBuilder.credentialsProvider(
-        V1V2AwsCredentialProviderAdapter.adapt(parameters.getCredentialSet()));
-
-    URI endpoint = getS3Endpoint(parameters.getEndpoint(), conf);
-
-    Region region =
-        getS3Region(conf.getTrimmed(AWS_REGION), parameters.getCredentialSet());
-
-    LOG.debug("Using endpoint {}; and region {}", endpoint, region);
-
-    s3ClientBuilder.endpointOverride(endpoint).region(region);
-
-    S3Configuration s3Configuration = S3Configuration.builder()
-        .pathStyleAccessEnabled(parameters.isPathStyleAccess())
-        .build();
-
-    s3ClientBuilder.serviceConfiguration(s3Configuration);
 
-    // TODO: Some configuration done in configureBasicParams is not done yet.
-    //  Need to verify how metrics collection can be done, as SDK V2 only
-    //  seems to have a metrics publisher.
-
-    return s3ClientBuilder.build();
+    return clientOverrideConfigBuilder.build();
   }
 
-
   /**
    * Create an {@link AmazonS3} client of type
    * {@link AmazonS3EncryptionV2} if CSE is enabled.
@@ -527,10 +551,12 @@ private static URI getS3Endpoint(String endpoint, final Configuration conf) {
    *
    * @param region AWS S3 Region set in the config. This property may not be set, in which case
    *               ask S3 for the region.
+   * @param bucket Bucket name.
    * @param credentialsProvider Credentials provider to be used with the default s3 client.
    * @return region of the bucket.
    */
-  private Region getS3Region(String region, AWSCredentialsProvider credentialsProvider) {
+  private static Region getS3Region(String region, String bucket,
+      AWSCredentialsProvider credentialsProvider) {
 
     if (!StringUtils.isBlank(region)) {
       return Region.of(region);

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

@@ -54,8 +54,6 @@
 import com.amazonaws.SdkBaseException;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.Headers;
-import com.amazonaws.services.s3.model.SelectObjectContentRequest;
-import com.amazonaws.services.s3.model.SelectObjectContentResult;
 import com.amazonaws.services.s3.transfer.TransferManager;
 import com.amazonaws.services.s3.transfer.TransferManagerConfiguration;
 
@@ -64,6 +62,8 @@
 import org.slf4j.LoggerFactory;
 
 import software.amazon.awssdk.core.ResponseInputStream;
+import software.amazon.awssdk.core.async.SdkPublisher;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
@@ -88,6 +88,10 @@
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.services.s3.model.PutObjectResponse;
 import software.amazon.awssdk.services.s3.model.S3Error;
+import software.amazon.awssdk.services.s3.model.SelectObjectContentEventStream;
+import software.amazon.awssdk.services.s3.model.SelectObjectContentRequest;
+import software.amazon.awssdk.services.s3.model.SelectObjectContentResponse;
+import software.amazon.awssdk.services.s3.model.SelectObjectContentResponseHandler;
 import software.amazon.awssdk.services.s3.model.StorageClass;
 import software.amazon.awssdk.services.s3.model.UploadPartRequest;
 import software.amazon.awssdk.services.s3.model.UploadPartResponse;
@@ -289,6 +293,7 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
   private String username;
   private AmazonS3 s3;
   private S3Client s3V2;
+  private S3AsyncClient s3AsyncClient;
   // initial callback policy is fail-once; it's there just to assist
   // some mock tests and other codepaths trying to call the low level
   // APIs on an uninitialized filesystem.
@@ -973,6 +978,9 @@ private void bindAWSClient(URI name, boolean dtEnabled) throws IOException {
         .createS3ClientV2(getUri(),
             parameters);
 
+    s3AsyncClient = ReflectionUtils.newInstance(s3ClientFactoryClass, conf)
+        .createS3AsyncClient(getUri(),
+            parameters);
   }
 
   /**
@@ -1689,8 +1697,10 @@ private final class WriteOperationHelperCallbacksImpl
       implements WriteOperationHelper.WriteOperationHelperCallbacks {
 
     @Override
-    public SelectObjectContentResult selectObjectContent(SelectObjectContentRequest request) {
-      return s3.selectObjectContent(request);
+    public CompletableFuture<Void> selectObjectContent(
+        SelectObjectContentRequest request,
+        SelectObjectContentResponseHandler responseHandler) {
+     return s3AsyncClient.selectObjectContent(request, responseHandler);
     }
 
     @Override
@@ -1700,7 +1710,6 @@ public CompleteMultipartUploadResponse completeMultipartUpload(
     }
   }
 
-
   /**
    * Create the read context for reading from the referenced file,
    * using FS state as well as the status.