Add dynamodb streams source (#6874)

### Motivation The goal is to allow consuming dynamodb streams directly into pulsar ### Modifications I created a new source for dynamo, which shares less code than ideal with the kinesis source, since the dynamodb kinesis client adapter supports KCL v1.x only, while, the kinesis source is using KCL v2.x. I also abstracted the aws credential management pieces into their own package. ### Verifying this change Create a dynamodb table with streams enabled. Configure this connector with the stream ARN and appropriate credentials. Create/update an entry in the table and ensure it is written to pulsar by the connector.
apache · May 19, 2020 · 812b8f2 · 812b8f2
1 parent 0bc1a6c
commit 812b8f2
Show file tree

Hide file tree

Showing 23 changed files with 1,211 additions and 152 deletions.
diff --git a/pulsar-io/aws/pom.xml b/pulsar-io/aws/pom.xml
@@ -0,0 +1,59 @@
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.apache.pulsar</groupId>
+    <artifactId>pulsar-io</artifactId>
+    <version>2.6.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>pulsar-io-aws</artifactId>
+  <name>Pulsar IO :: IO AWS</name>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.pulsar</groupId>
+      <artifactId>pulsar-io-core</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.google.code.gson</groupId>
+      <artifactId>gson</artifactId>
+    </dependency>
+
+    <!-- aws dependencies -->
+    <dependency>
+      <groupId>com.amazonaws</groupId>
+      <artifactId>aws-java-sdk-sts</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>sts</artifactId>
+      <version>2.10.56</version>
+    </dependency>
+	<!-- /aws dependencies -->
+
+  </dependencies>
+</project>
diff --git a/.../io/kinesis/AbstractKinesisConnector.java → ...e/pulsar/io/aws/AbstractAwsConnector.java b/.../io/kinesis/AbstractKinesisConnector.java → ...e/pulsar/io/aws/AbstractAwsConnector.java
@@ -16,10 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.io.kinesis;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static org.apache.commons.lang3.StringUtils.isNotBlank;
+package org.apache.pulsar.io.aws;
 
 import java.io.IOException;
 import java.lang.reflect.Constructor;
@@ -32,16 +29,17 @@
 import com.google.gson.reflect.TypeToken;
 
 import lombok.extern.slf4j.Slf4j;
+import software.amazon.awssdk.utils.StringUtils;
 
 @Slf4j
-public abstract class AbstractKinesisConnector {
+public abstract class AbstractAwsConnector {
 
     public static final String ACCESS_KEY_NAME = "accessKey";
     public static final String SECRET_KEY_NAME = "secretKey";
 
-    protected AwsCredentialProviderPlugin createCredentialProvider(String awsCredentialPluginName,
-            String awsCredentialPluginParam) {
-        if (isNotBlank(awsCredentialPluginName)) {
+    public AwsCredentialProviderPlugin createCredentialProvider(String awsCredentialPluginName,
+                                                                   String awsCredentialPluginParam) {
+        if (StringUtils.isNotBlank(awsCredentialPluginName)) {
             return createCredentialProviderWithPlugin(awsCredentialPluginName, awsCredentialPluginParam);
         } else {
             return defaultCredentialProvider(awsCredentialPluginParam);
@@ -79,16 +77,19 @@ public static AwsCredentialProviderPlugin createCredentialProviderWithPlugin(Str
      * @param awsCredentialPluginParam
      * @return
      */
-    protected AwsCredentialProviderPlugin defaultCredentialProvider(String awsCredentialPluginParam) {
+    public AwsCredentialProviderPlugin defaultCredentialProvider(String awsCredentialPluginParam) {
         Map<String, String> credentialMap = new Gson().fromJson(awsCredentialPluginParam,
                 new TypeToken<Map<String, String>>() {
                 }.getType());
         String accessKey = credentialMap.get(ACCESS_KEY_NAME);
         String secretKey = credentialMap.get(SECRET_KEY_NAME);
-        checkArgument(isNotBlank(accessKey) && isNotBlank(secretKey),
-                String.format(
-                        "Default %s and %s must be present into json-map if AwsCredentialProviderPlugin not provided",
-                        ACCESS_KEY_NAME, SECRET_KEY_NAME));
+        if (!(StringUtils.isNotBlank(accessKey) && StringUtils.isNotBlank(secretKey))) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Default %s and %s must be present into json-map if AwsCredentialProviderPlugin not provided",
+                            ACCESS_KEY_NAME, SECRET_KEY_NAME)
+            );
+        }
         return new AwsCredentialProviderPlugin() {
             @Override
             public void init(String param) {

diff --git a/pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/AwsCredentialProviderPlugin.java b/pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/AwsCredentialProviderPlugin.java
@@ -0,0 +1,76 @@
+/**
+ * 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.aws;
+
+import java.io.Closeable;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.AWSSessionCredentials;
+import com.amazonaws.auth.BasicSessionCredentials;
+
+/**
+ * Kinesis source/sink calls credential-provider while refreshing aws accessKey and secreKey. So, implementation
+ * AwsCredentialProviderPlugin needs to makes sure to return non-expired keys when it requires.
+ *
+ */
+public interface AwsCredentialProviderPlugin extends Closeable {
+
+    /**
+     * accepts aws-account related param and initialize credential provider.
+     * 
+     * @param param
+     */
+    void init(String param);
+
+    /**
+     * Returned {@link AWSCredentialsProvider} can give {@link AWSCredentials} in case credential belongs to IAM user or
+     * it can return {@link BasicSessionCredentials} if user wants to generate temporary credential for a given IAM
+     * role.
+     * 
+     * @return
+     */
+    AWSCredentialsProvider getCredentialProvider();
+
+    /**
+     * Returns a V2 credential provider for use with the v2 SDK.
+     *
+     * Defaults to an implementation that pulls credentials from a v1 provider
+     */
+    default software.amazon.awssdk.auth.credentials.AwsCredentialsProvider getV2CredentialsProvider() {
+        // make a small wrapper to forward requests to v1, this allows
+        // for this interface to not "break" for implementers
+        AWSCredentialsProvider v1Provider = getCredentialProvider();
+        return () -> {
+            AWSCredentials creds = v1Provider.getCredentials();
+            if (creds instanceof AWSSessionCredentials) {
+                return software.amazon.awssdk.auth.credentials.AwsSessionCredentials.create(
+                        creds.getAWSAccessKeyId(),
+                        creds.getAWSSecretKey(),
+                        ((AWSSessionCredentials) creds).getSessionToken());
+            } else {
+                return software.amazon.awssdk.auth.credentials.AwsBasicCredentials.create(
+                        creds.getAWSAccessKeyId(),
+                        creds.getAWSSecretKey());
+            }
+        };
+    }
+
+}
diff --git a/pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/AwsDefaultProviderChainPlugin.java b/pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/AwsDefaultProviderChainPlugin.java
@@ -0,0 +1,47 @@
+/**
+ * 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.aws;
+
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+
+import java.io.IOException;
+
+public class AwsDefaultProviderChainPlugin implements AwsCredentialProviderPlugin {
+    @Override
+    public void init(String param) {
+
+    }
+
+    @Override
+    public AWSCredentialsProvider getCredentialProvider() {
+        return new DefaultAWSCredentialsProviderChain();
+    }
+
+    @Override
+    public software.amazon.awssdk.auth.credentials.AwsCredentialsProvider getV2CredentialsProvider() {
+        return DefaultCredentialsProvider.create();
+    }
+
+    @Override
+    public void close() throws IOException {
+
+    }
+}
diff --git a/pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/STSAssumeRoleProviderPlugin.java b/pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/STSAssumeRoleProviderPlugin.java
@@ -0,0 +1,64 @@
+/**
+ * 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.aws;
+
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class STSAssumeRoleProviderPlugin implements AwsCredentialProviderPlugin {
+    public static final String ASSUME_ROLE_ARN = "roleArn";
+    public static final String ASSUME_ROLE_SESSION_NAME = "roleSessionName";
+
+    private String roleArn;
+    private String roleSessionName;
+
+    @Override
+    public void init(String param) {
+        Map<String, String> credentialMap = new Gson().fromJson(param,
+                new TypeToken<Map<String, String>>() {
+                }.getType());
+
+        roleArn = credentialMap.get(ASSUME_ROLE_ARN);
+        roleSessionName = credentialMap.get(ASSUME_ROLE_SESSION_NAME);
+    }
+
+    @Override
+    public AWSCredentialsProvider getCredentialProvider() {
+        return new STSAssumeRoleSessionCredentialsProvider.Builder(roleArn, roleSessionName).build();
+    }
+
+    @Override
+    public software.amazon.awssdk.auth.credentials.AwsCredentialsProvider getV2CredentialsProvider() {
+        StsClient client = StsClient.create();
+        return StsAssumeRoleCredentialsProvider.builder().stsClient(client).refreshRequest((req) -> {
+            req.roleArn(roleArn).roleSessionName(roleSessionName).build();
+        }).build();
+    }
+
+    @Override
+    public void close() throws IOException {
+    }
+}