[tests] Add Kinesis Sink integration tests (#42)

* [tests] Add Kinesis connector integration tests

* fixup

pulsar-io/aws/pom.xml

@@ -53,6 +53,7 @@
       <artifactId>sts</artifactId>
       <version>2.10.56</version>
     </dependency>
+
 	<!-- /aws dependencies -->
 
   </dependencies>

pulsar-io/kinesis/pom.xml

@@ -108,6 +108,18 @@
       <version>2.3.0</version>
     </dependency>
 
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>localstack</artifactId>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <scope>test</scope>
+    </dependency>
+
   </dependencies>
 
   <build>

pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/BaseKinesisConfig.java

@@ -38,13 +38,28 @@ public abstract class BaseKinesisConfig implements Serializable {
     )
     private String awsEndpoint = "";
 
+    @FieldDoc(
+            required = false,
+            defaultValue = "",
+            help = "Kinesis end-point port. It can be found at https://docs.aws.amazon.com/general/latest/gr/rande.html"
+    )
+    private Integer awsEndpointPort;
+
     @FieldDoc(
         required = false,
         defaultValue = "",
         help = "Appropriate aws region. E.g. us-west-1, us-west-2"
     )
     private String awsRegion = "";
 
+
+    @FieldDoc(
+            required = false,
+            defaultValue = "false",
+            help = "Tell to Kinesis Client to skip certificate validation. This is useful while performing local tests, it's recommended to always validate certificate in production environment."
+    )
+    private Boolean skipCertificateValidation = false;
+
     @FieldDoc(
         required = true,
         defaultValue = "",

pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java

@@ -164,10 +164,17 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc
 
         KinesisProducerConfiguration kinesisConfig = new KinesisProducerConfiguration();
         kinesisConfig.setKinesisEndpoint(kinesisSinkConfig.getAwsEndpoint());
+        if (kinesisSinkConfig.getAwsEndpointPort() != null) {
+            kinesisConfig.setKinesisPort(kinesisSinkConfig.getAwsEndpointPort());
+        }
         kinesisConfig.setRegion(kinesisSinkConfig.getAwsRegion());
         kinesisConfig.setThreadingModel(ThreadingModel.POOLED);
         kinesisConfig.setThreadPoolSize(4);
         kinesisConfig.setCollectionMaxCount(1);
+        if (kinesisSinkConfig.getSkipCertificateValidation() != null &&
+                kinesisSinkConfig.getSkipCertificateValidation()) {
+            kinesisConfig.setVerifyCertificate(false);
+        }
         AWSCredentialsProvider credentialsProvider = createCredentialProvider(
                 kinesisSinkConfig.getAwsCredentialPluginName(),
                 kinesisSinkConfig.getAwsCredentialPluginParam())

pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkAuthTest.java

@@ -0,0 +1,151 @@
+/**
+ * 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.pulsar.io.kinesis;
+
+import java.io.IOException;
+import java.util.Map;
+
+import org.apache.pulsar.io.aws.AwsCredentialProviderPlugin;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import org.testng.collections.Maps;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.BasicSessionCredentials;
+import com.google.gson.Gson;
+
+public class KinesisSinkAuthTest {
+
+    @Test
+    public void testDefaultCredentialProvider() throws Exception {
+        KinesisSink sink = new KinesisSink();
+        Map<String, String> credentialParam = Maps.newHashMap();
+        String awsCredentialPluginParam = new Gson().toJson(credentialParam);
+        try {
+            sink.defaultCredentialProvider(awsCredentialPluginParam);
+            Assert.fail("accessKey and SecretKey validation not applied");
+        } catch (IllegalArgumentException ie) {
+            // Ok..
+        }
+
+        final String accesKey = "ak";
+        final String secretKey = "sk";
+        credentialParam.put(KinesisSink.ACCESS_KEY_NAME, accesKey);
+        credentialParam.put(KinesisSink.SECRET_KEY_NAME, secretKey);
+        awsCredentialPluginParam = new Gson().toJson(credentialParam);
+        AWSCredentialsProvider credentialProvider = sink.defaultCredentialProvider(awsCredentialPluginParam)
+                .getCredentialProvider();
+        Assert.assertNotNull(credentialProvider);
+        Assert.assertEquals(credentialProvider.getCredentials().getAWSAccessKeyId(), accesKey);
+        Assert.assertEquals(credentialProvider.getCredentials().getAWSSecretKey(), secretKey);
+
+        sink.close();
+    }
+
+    @Test
+    public void testCredentialProvider() throws Exception {
+        KinesisSink sink = new KinesisSink();
+
+        final String accesKey = "ak";
+        final String secretKey = "sk";
+        Map<String, String> credentialParam = Maps.newHashMap();
+        credentialParam.put(KinesisSink.ACCESS_KEY_NAME, accesKey);
+        credentialParam.put(KinesisSink.SECRET_KEY_NAME, secretKey);
+        String awsCredentialPluginParam = new Gson().toJson(credentialParam);
+        AWSCredentialsProvider credentialProvider = sink.createCredentialProvider(null, awsCredentialPluginParam)
+                .getCredentialProvider();
+        Assert.assertEquals(credentialProvider.getCredentials().getAWSAccessKeyId(), accesKey);
+        Assert.assertEquals(credentialProvider.getCredentials().getAWSSecretKey(), secretKey);
+
+        credentialProvider = sink.createCredentialProvider(AwsCredentialProviderPluginImpl.class.getName(), "{}")
+                .getCredentialProvider();
+        Assert.assertNotNull(credentialProvider);
+        Assert.assertEquals(credentialProvider.getCredentials().getAWSAccessKeyId(),
+                AwsCredentialProviderPluginImpl.accessKey);
+        Assert.assertEquals(credentialProvider.getCredentials().getAWSSecretKey(),
+                AwsCredentialProviderPluginImpl.secretKey);
+        Assert.assertEquals(((BasicSessionCredentials) credentialProvider.getCredentials()).getSessionToken(),
+                AwsCredentialProviderPluginImpl.sessionToken);
+
+        sink.close();
+    }
+
+    @Test
+    public void testCredentialProviderPlugin() throws Exception {
+        KinesisSink sink = new KinesisSink();
+
+        AWSCredentialsProvider credentialProvider = sink
+                .createCredentialProviderWithPlugin(AwsCredentialProviderPluginImpl.class.getName(), "{}")
+                .getCredentialProvider();
+        Assert.assertNotNull(credentialProvider);
+        Assert.assertEquals(credentialProvider.getCredentials().getAWSAccessKeyId(),
+                AwsCredentialProviderPluginImpl.accessKey);
+        Assert.assertEquals(credentialProvider.getCredentials().getAWSSecretKey(),
+                AwsCredentialProviderPluginImpl.secretKey);
+        Assert.assertEquals(((BasicSessionCredentials) credentialProvider.getCredentials()).getSessionToken(),
+                AwsCredentialProviderPluginImpl.sessionToken);
+
+        sink.close();
+    }
+
+    public static class AwsCredentialProviderPluginImpl implements AwsCredentialProviderPlugin {
+
+        public final static String accessKey = "ak";
+        public final static String secretKey = "sk";
+        public final static String sessionToken = "st";
+
+        public void init(String param) {
+            // no-op
+        }
+
+        @Override
+        public AWSCredentialsProvider getCredentialProvider() {
+            return new AWSCredentialsProvider() {
+                @Override
+                public AWSCredentials getCredentials() {
+                    return new BasicSessionCredentials(accessKey, secretKey, sessionToken) {
+
+                        @Override
+                        public String getAWSAccessKeyId() {
+                            return accessKey;
+                        }
+                        @Override
+                        public String getAWSSecretKey() {
+                            return secretKey;
+                        }
+                        @Override
+                        public String getSessionToken() {
+                            return sessionToken;
+                        }
+                    };
+                }
+                @Override
+                public void refresh() {
+                    // TODO Auto-generated method stub
+                }
+            };
+        }
+        @Override
+        public void close() throws IOException {
+            // TODO Auto-generated method stub
+        }
+    }
+
+}