Add testing changes

remove validations

Author: khsoneji

kafka-connect-s3/src/main/java/io/confluent/connect/s3/S3SinkConnectorConfig.java

@@ -75,11 +75,14 @@
 import io.confluent.connect.storage.partitioner.HourlyPartitioner;
 import io.confluent.connect.storage.partitioner.PartitionerConfig;
 import io.confluent.connect.storage.partitioner.TimeBasedPartitioner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
 
 public class S3SinkConnectorConfig extends StorageSinkConnectorConfig {
 
+  private static final Logger log = LoggerFactory.getLogger(S3SinkConnectorConfig.class);
   // S3 Group
   public static final String S3_BUCKET_CONFIG = "s3.bucket.name";
 
@@ -949,8 +952,8 @@ public String awsMiddlewareRoleARN() {
     return getString(MIDDLEWARE_ROLE_ARN_CONFIG);
   }
 
-  public String awsExternalId() {
-    return getString(CUSTOMER_ROLE_EXTERNAL_ID_CONFIG);
+  public Password awsExternalId() {
+    return getPassword(CUSTOMER_ROLE_EXTERNAL_ID_CONFIG);
   }
 
   public int getPartSize() {
@@ -966,32 +969,37 @@ public AWSCredentialsProvider getCredentialsProvider() {
     try {
       AWSCredentialsProvider provider = ((Class<? extends AWSCredentialsProvider>)
           getClass(S3SinkConnectorConfig.CREDENTIALS_PROVIDER_CLASS_CONFIG)).newInstance();
+      
+      String authMethod = getAuthenticationMethod();
+      log.info("Authentication method: {}", authMethod);
 
       if (provider instanceof Configurable) {
         Map<String, Object> configs = originalsWithPrefix(CREDENTIALS_PROVIDER_CONFIG_PREFIX);
         configs.remove(CREDENTIALS_PROVIDER_CLASS_CONFIG.substring(
             CREDENTIALS_PROVIDER_CONFIG_PREFIX.length()
         ));
 
-        String authMethod = getAuthenticationMethod();
-        if (authMethod == "IAM Assume Role") {
+        configs.put(AWS_ACCESS_KEY_ID_CONFIG, awsAccessKeyId());
+        configs.put(AWS_SECRET_ACCESS_KEY_CONFIG, awsSecretKeyId().value());
+        ((Configurable) provider).configure(configs);
+      } else {
+        authMethod = getAuthenticationMethod();
+
+        if (authMethod.equals("IAM Assume Role")) {
+          Map<String, Object> configs = new HashMap<String, Object>();
           configs.put(CUSTOMER_ROLE_ARN_CONFIG, awsCustomerRoleARN());
-          configs.put(CUSTOMER_ROLE_EXTERNAL_ID_CONFIG, awsExternalId());
+          configs.put(CUSTOMER_ROLE_EXTERNAL_ID_CONFIG, awsExternalId().value());
           configs.put(MIDDLEWARE_ROLE_ARN_CONFIG, awsMiddlewareRoleARN());
 
           provider = new AwsIamAssumeRoleChaining();
           ((AwsIamAssumeRoleChaining) provider).configure(configs);
         } else {
-          configs.put(AWS_ACCESS_KEY_ID_CONFIG, awsAccessKeyId());
-          configs.put(AWS_SECRET_ACCESS_KEY_CONFIG, awsSecretKeyId().value());
-          ((Configurable) provider).configure(configs);
-        }
-      } else {
-        final String accessKeyId = awsAccessKeyId();
-        final String secretKey = awsSecretKeyId().value();
-        if (StringUtils.isNotBlank(accessKeyId) && StringUtils.isNotBlank(secretKey)) {
-          BasicAWSCredentials basicCredentials = new BasicAWSCredentials(accessKeyId, secretKey);
-          provider = new AWSStaticCredentialsProvider(basicCredentials);
+          final String accessKeyId = awsAccessKeyId();
+          final String secretKey = awsSecretKeyId().value();
+          if (StringUtils.isNotBlank(accessKeyId) && StringUtils.isNotBlank(secretKey)) {
+            BasicAWSCredentials basicCredentials = new BasicAWSCredentials(accessKeyId, secretKey);
+            provider = new AWSStaticCredentialsProvider(basicCredentials);
+          }
         }
       }
 

kafka-connect-s3/src/main/java/io/confluent/connect/s3/auth/iamassume/AwsIamAssumeRoleChaining.java

@@ -21,8 +21,11 @@
 import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider;
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClientBuilder;
+
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import static io.confluent.connect.s3.S3SinkConnectorConfig.CUSTOMER_ROLE_ARN_CONFIG;
 import static io.confluent.connect.s3.S3SinkConnectorConfig.CUSTOMER_ROLE_EXTERNAL_ID_CONFIG;
@@ -31,11 +34,12 @@
 
 public class AwsIamAssumeRoleChaining implements AWSCredentialsProvider {
 
+  private static final Logger log = LoggerFactory.getLogger(AwsIamAssumeRoleChaining.class);
   private static final ConfigDef STS_CONFIG_DEF = new ConfigDef()
       .define(
         CUSTOMER_ROLE_EXTERNAL_ID_CONFIG,
         ConfigDef.Type.STRING,
-        ConfigDef.Importance.MEDIUM,
+        ConfigDef.Importance.HIGH,
         "The role external ID used when retrieving session credentials under an assumed role."
       ).define(
         CUSTOMER_ROLE_ARN_CONFIG,
@@ -53,6 +57,7 @@ public class AwsIamAssumeRoleChaining implements AWSCredentialsProvider {
   private String customerRoleExternalId;
   private String middlewareRoleArn;
   private STSAssumeRoleSessionCredentialsProvider stsCredentialProvider;
+  private STSAssumeRoleSessionCredentialsProvider initialProvider;
 
   // Method to initiate role chaining
   public void configure(Map<String, ?> configs) {
@@ -62,12 +67,13 @@ public void configure(Map<String, ?> configs) {
     customerRoleExternalId = config.getString(CUSTOMER_ROLE_EXTERNAL_ID_CONFIG);
     middlewareRoleArn = config.getString(MIDDLEWARE_ROLE_ARN_CONFIG);
 
-    STSAssumeRoleSessionCredentialsProvider initialProvider = buildProvider(
+    initialProvider = buildProvider(
         middlewareRoleArn, 
         "middlewareSession", 
         "", 
         null
     );
+    log.info("Got initial provider");
 
     // Use the credentials from the initial role to assume the subsequent role
     stsCredentialProvider = buildProvider(
@@ -76,6 +82,7 @@ public void configure(Map<String, ?> configs) {
         customerRoleExternalId, 
         initialProvider
     );
+    log.info("Got final credentials");
   }
 
   // Updated buildProvider to optionally accept an existing AwsCredentialsProvider
@@ -99,10 +106,9 @@ private STSAssumeRoleSessionCredentialsProvider buildProvider(
           .build();
     } else {
       credentialsProvider = new STSAssumeRoleSessionCredentialsProvider
-          .Builder(roleArn, roleSessionName)
-          .withStsClient(AWSSecurityTokenServiceClientBuilder.defaultClient())
-          .withExternalId(roleExternalId)
-          .build();
+        .Builder(roleArn, roleSessionName)
+        .withStsClient(AWSSecurityTokenServiceClientBuilder.defaultClient())
+        .build();
     }
     return credentialsProvider;
   }
@@ -114,6 +120,15 @@ public AWSCredentials getCredentials() {
 
   @Override
   public void refresh() {
-    stsCredentialProvider.refresh();
+    if (initialProvider != null) {
+      initialProvider.refresh();
+      stsCredentialProvider = buildProvider(
+          customerRoleArn, 
+          "customerSession", 
+          customerRoleExternalId, 
+          initialProvider
+      );
+    }
+    //stsCredentialProvider.refresh();
   }
 }