Support protocol configuration for S3 (#6954)

* Support protocol configuration for S3

* Add doc

aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java

@@ -24,6 +24,9 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 
 public class AWSClientConfig
 {
+  @JsonProperty
+  private String protocol = "https"; // The default of aws-java-sdk
+
   @JsonProperty
   private boolean disableChunkedEncoding = S3ClientOptions.DEFAULT_CHUNKED_ENCODING_DISABLED;
 
@@ -33,6 +36,11 @@ public class AWSClientConfig
   @JsonProperty
   protected boolean forceGlobalBucketAccessEnabled = S3ClientOptions.DEFAULT_FORCE_GLOBAL_BUCKET_ACCESS_ENABLED;
 
+  public String getProtocol()
+  {
+    return protocol;
+  }
+
   public boolean isDisableChunkedEncoding()
   {
     return disableChunkedEncoding;

docs/content/development/extensions-core/s3.md

@@ -48,6 +48,7 @@ As an example, to set the region to 'us-east-1' through system properties:
 |`druid.storage.sse.type`|Server-side encryption type. Should be one of `s3`, `kms`, and `custom`. See the below [Server-side encryption section](#server-side-encryption) for more details.|None|
 |`druid.storage.sse.kms.keyId`|AWS KMS key ID. Can be empty if `druid.storage.sse.type` is `kms`.|None|
 |`druid.storage.sse.custom.base64EncodedKey`|Base64-encoded key. Should be specified if `druid.storage.sse.type` is `custom`.|None|
+|`druid.s3.protocol`|Communication protocol type to use when sending requests to AWS. `http` or `https` can be used.|`https`|
 |`druid.s3.disableChunkedEncoding`|Disables chunked encoding. See [AWS document](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3Builder.html#disableChunkedEncoding--) for details.|false|
 |`druid.s3.enablePathStyleAccess`|Enables path style access. See [AWS document](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3Builder.html#enablePathStyleAccess--) for details.|false|
 |`druid.s3.forceGlobalBucketAccessEnabled`|Enables global bucket access. See [AWS document](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3Builder.html#setForceGlobalBucketAccessEnabled-java.lang.Boolean-) for details.|false|

extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3StorageDruidModule.java

@@ -21,6 +21,7 @@ package org.apache.druid.storage.s3;
 
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.ClientConfigurationFactory;
+import com.amazonaws.Protocol;
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration;
 import com.amazonaws.services.s3.AmazonS3Client;
@@ -41,7 +42,11 @@ import org.apache.druid.guice.Binders;
 import org.apache.druid.guice.JsonConfigProvider;
 import org.apache.druid.guice.LazySingleton;
 import org.apache.druid.initialization.DruidModule;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.logger.Logger;
 
+import javax.annotation.Nullable;
+import java.net.URI;
 import java.util.List;
 
 /**
@@ -50,6 +55,8 @@ public class S3StorageDruidModule implements DruidModule
 {
   public static final String SCHEME = "s3_zip";
 
+  private static final Logger log = new Logger(S3StorageDruidModule.class);
+
   @Override
   public List<? extends Module> getJacksonModules()
   {
@@ -122,10 +129,11 @@ public class S3StorageDruidModule implements DruidModule
   )
   {
     final ClientConfiguration configuration = new ClientConfigurationFactory().getConfig();
+    final Protocol protocol = determineProtocol(clientConfig, endpointConfig);
     final AmazonS3ClientBuilder builder = AmazonS3Client
         .builder()
         .withCredentials(provider)
-        .withClientConfiguration(setProxyConfig(configuration, proxyConfig))
+        .withClientConfiguration(setProxyConfig(configuration, proxyConfig).withProtocol(protocol))
         .withChunkedEncodingDisabled(clientConfig.isDisableChunkedEncoding())
         .withPathStyleAccessEnabled(clientConfig.isEnablePathStyleAccess())
         .withForceGlobalBucketAccessEnabled(clientConfig.isForceGlobalBucketAccessEnabled());
@@ -136,10 +144,7 @@ public class S3StorageDruidModule implements DruidModule
       );
     }
 
-    return new ServerSideEncryptingAmazonS3(
+    return new ServerSideEncryptingAmazonS3(builder.build(), storageConfig.getServerSideEncryption());
-        builder.build(),
-        storageConfig.getServerSideEncryption()
-    );
   }
 
   private static ClientConfiguration setProxyConfig(ClientConfiguration conf, AWSProxyConfig proxyConfig)
@@ -158,4 +163,36 @@ public class S3StorageDruidModule implements DruidModule
     }
     return conf;
   }
+
+  @Nullable
+  private static Protocol parseProtocol(@Nullable String protocol)
+  {
+    if (protocol == null) {
+      return null;
+    }
+
+    if (protocol.equalsIgnoreCase("http")) {
+      return Protocol.HTTP;
+    } else if (protocol.equalsIgnoreCase("https")) {
+      return Protocol.HTTPS;
+    } else {
+      throw new IAE("Unknown protocol[%s]", protocol);
+    }
+  }
+
+  private static Protocol determineProtocol(AWSClientConfig clientConfig, AWSEndpointConfig endpointConfig)
+  {
+    final Protocol protocolFromClientConfig = parseProtocol(clientConfig.getProtocol());
+    final String endpointUrl = endpointConfig.getUrl();
+    if (StringUtils.isNotEmpty(endpointUrl)) {
+      final URI uri = URI.create(endpointUrl);
+      final Protocol protocol = parseProtocol(uri.getScheme());
+      if (protocol != null && (protocol != protocolFromClientConfig)) {
+        log.warn("[%s] protocol will be used for endpoint [%s]", protocol, endpointUrl);
+      }
+      return protocol;
+    } else {
+      return protocolFromClientConfig;
+    }
+  }
 }