INS-34515-2 (#18)

* INS-34515-2

* INS-34515-2

* Log error message from provider

* Bugfix typo

* Fixes for getting to talk from ontap

* Add enablePathStyle option

* Bugfix

* Regression bugfixes

* Rename classes after testing

* Extra logging to figure out validating errors

* Version bump

* Codereview unnecessary check

* Rename files

Author: niluka-insta

distribution.xml

@@ -12,13 +12,13 @@
 
     <artifactId>distribution</artifactId>
     <packaging>pom</packaging>
-    <version>0.1.4</version>
+    <version>0.1.5</version>
 
     <dependencies>
         <dependency>
             <groupId>com.instaclustr.kafkaconnect</groupId>
             <artifactId>instaclustr-s3-connector</artifactId>
-            <version>0.1.4</version>
+            <version>0.1.5</version>
         </dependency>
         <dependency>
             <groupId>com.instaclustr.kafkaconnect</groupId>

s3/pom.xml

@@ -10,7 +10,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>instaclustr-s3-connector</artifactId>
-    <version>0.1.4</version>
+    <version>0.1.5</version>
     <packaging>jar</packaging>
 
     <dependencies>

s3/src/main/java/com/instaclustr/kafka/connect/s3/AwsStorageConnectorCommonConfig.java

@@ -5,10 +5,13 @@
 import com.amazonaws.services.s3.model.AmazonS3Exception;
 import com.amazonaws.services.s3.model.Region;
 import com.amazonaws.services.securitytoken.model.AWSSecurityTokenServiceException;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.common.config.Config;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigValue;
 import org.apache.kafka.connect.errors.ConnectException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 import java.util.regex.Pattern;
@@ -19,18 +22,16 @@
 
 public class AwsStorageConnectorCommonConfig {
     public static final String BUCKET = "aws.s3.bucket";
-
     public static final String AWS_REGION = "aws.region";
-
     public static final String S3_KEY_PREFIX = "prefix";
-
     public static final String AWS_SECRET_KEY = "aws.secretKey";
-
     public static final String AWS_ACCESS_KEY_ID = "aws.accessKeyId";
-
+    public static final String S3_ENDPOINT = "s3.endpoint";
     public static final String AWS_IAM_ROLE_ARN = "aws.role.arn";
+    public static final String S3_ENABLE_PATH_STYLE = "s3.enablePathStyle";
 
     public static final String DEFAULT_AWS_REGION = Regions.DEFAULT_REGION.getName();
+    private static final Logger logger = LoggerFactory.getLogger(AwsStorageConnectorCommonConfig.class);
 
     private AwsStorageConnectorCommonConfig() {}
 
@@ -42,7 +43,9 @@ public static ConfigDef conf() {
                 .define(AWS_ACCESS_KEY_ID, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "AWS access key id")
                 .define(AWS_SECRET_KEY, ConfigDef.Type.PASSWORD, ConfigDef.Importance.HIGH, "AWS access secret key")
                 .define(AWS_REGION, ConfigDef.Type.STRING, DEFAULT_AWS_REGION, ConfigDef.Importance.MEDIUM, String.format("AWS client region, if not set will use %s", DEFAULT_AWS_REGION))
-                .define(AWS_IAM_ROLE_ARN, ConfigDef.Type.STRING, "", ConfigDef.Importance.HIGH, "");
+                .define(AWS_IAM_ROLE_ARN, ConfigDef.Type.STRING, "", ConfigDef.Importance.HIGH, "")
+                .define(S3_ENDPOINT, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, "Optional, S3 endpoint URL used to make it compatible with certain storage endpoints")
+                .define(S3_ENABLE_PATH_STYLE, ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.MEDIUM, "Optional, ensures the bucket name is in the URL path, making it compatible with certain storage endpoints");
         return configDef;
     }
 
@@ -61,7 +64,7 @@ public static void verifyS3CredentialsAndBucketInfo(final Map<String, String> se
             String awsRegion = sentConfigMap.get(AWS_REGION);
             AmazonS3 s3Client = TransferManagerProvider.getS3ClientBuilderWithRegionAndCredentials(sentConfigMap).build();
             if (s3Client.doesBucketExistV2(s3BucketName)) {
-                if (awsRegion != null) {
+                if (StringUtils.isBlank(sentConfigMap.get(S3_ENDPOINT)) && awsRegion != null) {
                     String bucketRegion = Region.fromValue(s3Client.getBucketLocation(s3BucketName)).toAWSRegion().getName();
                     if (!bucketRegion.equals(awsRegion)) {
                         addErrorMessageToConfigObject(configObject, AWS_REGION, String.format("Defined region(%s) is not the same as the bucket region(%s)", awsRegion, bucketRegion));
@@ -95,7 +98,8 @@ public static void verifyS3CredentialsAndBucketInfo(final Map<String, String> se
                     throw new ConnectException(String.format("Unknown Amazon S3 exception while validating config, %s", e.getErrorCode()), e);
             }
         } catch (IllegalArgumentException e) {
-            addErrorMessageToConfigObject(configObject, AWS_REGION, "The defined aws.region is invalid");
+            logger.info("Error whilst validating configurations, {}", e.getMessage());
+            addErrorMessageToConfigObject(configObject, AWS_REGION, String.format("The defined aws.region is invalid %s", e.getMessage()));
         }
     }
 

s3/src/main/java/com/instaclustr/kafka/connect/s3/TransferManagerProvider.java

@@ -5,6 +5,7 @@
 import com.amazonaws.auth.AWSStaticCredentialsProvider;
 import com.amazonaws.auth.BasicAWSCredentials;
 import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider;
+import com.amazonaws.client.builder.AwsClientBuilder;
 import com.amazonaws.regions.Regions;
 import com.amazonaws.services.s3.AmazonS3ClientBuilder;
 import com.amazonaws.services.s3.transfer.TransferManager;
@@ -17,7 +18,6 @@
 
 import java.util.Map;
 import java.util.UUID;
-import java.util.concurrent.TimeUnit;
 
 public class TransferManagerProvider {
     private TransferManager transferManager;
@@ -34,10 +34,11 @@ public TransferManagerProvider(final Map<String, String> config) {
     }
 
     public static AmazonS3ClientBuilder getS3ClientBuilderWithRegionAndCredentials(final Map<String, String> config) {
-        String accessKey = getFromConfigOrEnvironment(config, AwsStorageConnectorCommonConfig.AWS_ACCESS_KEY_ID);
-        String secret = getFromConfigOrEnvironment(config, AwsStorageConnectorCommonConfig.AWS_SECRET_KEY);
+        final String accessKey = getFromConfigOrEnvironment(config, AwsStorageConnectorCommonConfig.AWS_ACCESS_KEY_ID);
+        final String secret = getFromConfigOrEnvironment(config, AwsStorageConnectorCommonConfig.AWS_SECRET_KEY);
         String region = getFromConfigOrEnvironment(config, AwsStorageConnectorCommonConfig.AWS_REGION);
-        String roleArn = getFromConfigOrEnvironment(config, AwsStorageConnectorCommonConfig.AWS_IAM_ROLE_ARN);
+        final String roleArn = getFromConfigOrEnvironment(config, AwsStorageConnectorCommonConfig.AWS_IAM_ROLE_ARN);
+        final String endpoint = getFromConfigOrEnvironment(config, AwsStorageConnectorCommonConfig.S3_ENDPOINT);
 
         AWSStaticCredentialsProvider awsStaticCredentialsProvider = new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKey, secret));
         AWSCredentialsProvider awsCredentialsProvider;
@@ -64,12 +65,21 @@ public static AmazonS3ClientBuilder getS3ClientBuilderWithRegionAndCredentials(f
         AmazonS3ClientBuilder clientBuilder = AmazonS3ClientBuilder.standard()
                 .withCredentials(awsCredentialsProvider);
 
-        if (region == null) {
+        if (region == null && StringUtils.isBlank(endpoint)) {
             region = AwsStorageConnectorCommonConfig.DEFAULT_AWS_REGION;
             clientBuilder.enableForceGlobalBucketAccess();
             log.info("No region defined. Using {} and force global bucket access", AwsStorageConnectorCommonConfig.DEFAULT_AWS_REGION);
         }
-        clientBuilder.withRegion(Regions.fromName(region).getName()); //using fromName to validate the region value
+
+        if (StringUtils.isNotBlank(endpoint)) {
+            final boolean isPathStyleAccessEnabled = Boolean.parseBoolean(getFromConfigOrEnvironment(config, AwsStorageConnectorCommonConfig.S3_ENABLE_PATH_STYLE));
+
+            AwsClientBuilder.EndpointConfiguration endpointConfiguration = new AwsClientBuilder.EndpointConfiguration(endpoint, region);
+            clientBuilder.withEndpointConfiguration(endpointConfiguration).withPathStyleAccessEnabled(isPathStyleAccessEnabled);
+        } else {
+            clientBuilder.withRegion(Regions.fromName(region).getName()); //using fromName to validate the region value
+        }
+
         return clientBuilder;
     }
 

s3/src/main/java/com/instaclustr/kafka/connect/s3/source/AwsStorageSourceTask.java

@@ -3,7 +3,6 @@
 import com.amazonaws.AmazonClientException;
 import com.google.common.util.concurrent.RateLimiter;
 import com.google.common.util.concurrent.UncheckedExecutionException;
-import com.google.common.util.concurrent.UncheckedTimeoutException;
 import com.instaclustr.kafka.connect.s3.AwsConnectorStringFormats;
 import com.instaclustr.kafka.connect.s3.AwsStorageConnectorCommonConfig;
 import com.instaclustr.kafka.connect.s3.TransferManagerProvider;