🎉 New Destination: DynamoDB (#5561)

* Added the DynamoDB destination connector.
Implemented getConsumer and check methods.

Signed-off-by: Jinni Gu <jinnigu@uw.edu>

* Added auto-generated project files.

Signed-off-by: Yiqing Wang <yiqing@wangemail.com>

* Added config related files and output table helper.

Signed-off-by: Yiqing Wang <yiqing@wangemail.com>

* Added document for DynamoDB destination.

Signed-off-by: Jinni Gu <jinnigu@uw.edu>

* Implemented DynamodbWriter.
Added integration tests and unit tests.

Signed-off-by: qtz123 <qiutingzhi1995@gmail.com>

* Added DynamoDB in the SUMMARY.md.

Signed-off-by: qtz123 <qiutingzhi1995@gmail.com>

* Formatted code using ./gradlew format.

Signed-off-by: Jinni Gu <jinnigu@uw.edu>

* Added changelog to the doc.

Signed-off-by: qtz123 <qiutingzhi1995@gmail.com>

* Used PAY_PER_REQUEST instead of provisioned for DynamoDB.
Gave the value a name batchSize.
Removed unnecessary logs.

Signed-off-by: Yiqing Wang <yiqing@wangemail.com>

Co-authored-by: Yiqing Wang <yiqing@wangemail.com>
Co-authored-by: qtz123 <qiutingzhi1995@gmail.com>

airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/8ccd8909-4e99-4141-b48d-4984b70b2d89.json

@@ -0,0 +1,7 @@
+{
+  "destinationDefinitionId": "8ccd8909-4e99-4141-b48d-4984b70b2d89",
+  "name": "DynamoDB",
+  "dockerRepository": "airbyte/destination-dynamodb",
+  "dockerImageTag": "0.1.0",
+  "documentationUrl": "https://docs.airbyte.io/integrations/destinations/dynamodb"
+}

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -80,6 +80,11 @@
   dockerRepository: airbyte/destination-kafka
   dockerImageTag: 0.1.1
   documentationUrl: https://docs.airbyte.io/integrations/destinations/kafka
+- destinationDefinitionId: 8ccd8909-4e99-4141-b48d-4984b70b2d89
+  name: DynamoDB
+  dockerRepository: airbyte/destination-dynamodb
+  dockerImageTag: 0.1.0
+  documentationUrl: https://docs.airbyte.io/integrations/destinations/dynamodb
 - destinationDefinitionId: 8aaf41d0-f6d2-46de-9e79-c9540f828142
   name: Keen
   dockerRepository: airbyte/destination-keen

airbyte-integrations/connectors/destination-dynamodb/.dockerignore

@@ -0,0 +1,3 @@
+*
+!Dockerfile
+!build

airbyte-integrations/connectors/destination-dynamodb/Dockerfile

@@ -0,0 +1,11 @@
+FROM airbyte/integration-base-java:dev
+
+WORKDIR /airbyte
+ENV APPLICATION destination-dynamodb
+
+COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
+
+RUN tar xf ${APPLICATION}.tar --strip-components=1
+
+LABEL io.airbyte.version=0.1.0
+LABEL io.airbyte.name=airbyte/destination-dynamodb

airbyte-integrations/connectors/destination-dynamodb/README.md

@@ -0,0 +1,68 @@
+# Destination Dynamodb
+
+This is the repository for the Dynamodb destination connector in Java.
+For information about how to use this connector within Airbyte, see [the User Documentation](https://docs.airbyte.io/integrations/destinations/dynamodb).
+
+## Local development
+
+#### Building via Gradle
+From the Airbyte repository root, run:
+```
+./gradlew :airbyte-integrations:connectors:destination-dynamodb:build
+```
+
+#### Create credentials
+**If you are a community contributor**, generate the necessary credentials and place them in `secrets/config.json` conforming to the spec file in `src/main/resources/spec.json`.
+Note that the `secrets` directory is git-ignored by default, so there is no danger of accidentally checking in sensitive information.
+
+**If you are an Airbyte core member**, follow the [instructions](https://docs.airbyte.io/connector-development#using-credentials-in-ci) to set up the credentials.
+
+### Locally running the connector docker image
+
+#### Build
+Build the connector image via Gradle:
+```
+./gradlew :airbyte-integrations:connectors:destination-dynamodb:airbyteDocker
+```
+When building via Gradle, the docker image name and tag, respectively, are the values of the `io.airbyte.name` and `io.airbyte.version` `LABEL`s in
+the Dockerfile.
+
+#### Run
+Then run any of the connector commands as follows:
+```
+docker run --rm airbyte/destination-dynamodb:dev spec
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/destination-dynamodb:dev check --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/destination-dynamodb:dev discover --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets -v $(pwd)/integration_tests:/integration_tests airbyte/destination-dynamodb:dev read --config /secrets/config.json --catalog /integration_tests/configured_catalog.json
+```
+
+## Testing
+We use `JUnit` for Java tests.
+
+### Unit and Integration Tests
+Place unit tests under `src/test/io/airbyte/integrations/destinations/dynamodb`.
+
+#### Acceptance Tests
+Airbyte has a standard test suite that all destination connectors must pass. Implement the `TODO`s in
+`src/test-integration/java/io/airbyte/integrations/destinations/dynamodbDestinationAcceptanceTest.java`.
+
+### Using gradle to run tests
+All commands should be run from airbyte project root.
+To run unit tests:
+```
+./gradlew :airbyte-integrations:connectors:destination-dynamodb:unitTest
+```
+To run acceptance and custom integration tests:
+```
+./gradlew :airbyte-integrations:connectors:destination-dynamodb:integrationTest
+```
+
+## Dependency Management
+
+### Publishing a new version of the connector
+You've checked out the repo, implemented a million dollar feature, and you're ready to share your changes with the world. Now what?
+1. Make sure your changes are passing unit and integration tests.
+1. Bump the connector version in `Dockerfile` -- just increment the value of the `LABEL io.airbyte.version` appropriately (we use [SemVer](https://semver.org/)).
+1. Create a Pull Request.
+1. Pat yourself on the back for being an awesome contributor.
+1. Someone from Airbyte will take a look at your PR and iterate with you to merge it into master.

airbyte-integrations/connectors/destination-dynamodb/build.gradle

@@ -0,0 +1,21 @@
+plugins {
+    id 'application'
+    id 'airbyte-docker'
+    id 'airbyte-integration-test-java'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.destination.dynamodb.DynamodbDestination'
+}
+
+dependencies {
+    implementation project(':airbyte-config:models')
+    implementation project(':airbyte-protocol:models')
+    implementation project(':airbyte-integrations:bases:base-java')
+    implementation project(':airbyte-integrations:connectors:destination-jdbc')
+    implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
+    implementation 'com.amazonaws:aws-java-sdk-dynamodb:1.12.47'
+
+    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
+    integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-dynamodb')
+}

airbyte-integrations/connectors/destination-dynamodb/sample_secrets/config.json

@@ -0,0 +1,6 @@
+{
+  "dynamodb_table_name": "paste-table-name-here",
+  "dynamodb_region": "paste-dynamodb-region-here",
+  "access_key_id": "paste-access-key-id-here",
+  "secret_access_key": "paste-secret-access-key-here"
+}

airbyte-integrations/connectors/destination-dynamodb/src/main/java/io/airbyte/integrations/destination/dynamodb/DynamodbChecker.java

@@ -0,0 +1,100 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.integrations.destination.dynamodb;
+
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
+import com.amazonaws.services.dynamodbv2.document.*;
+import com.amazonaws.services.dynamodbv2.model.*;
+import io.airbyte.integrations.base.JavaBaseConstants;
+import java.util.Arrays;
+import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DynamodbChecker {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(DynamodbChecker.class);
+
+  public static void attemptDynamodbWriteAndDelete(DynamodbDestinationConfig dynamodbDestinationConfig) throws Exception {
+    var prefix = dynamodbDestinationConfig.getTableName();
+    final String outputTableName = prefix + "_airbyte_connection_test_" + UUID.randomUUID().toString().replaceAll("-", "");
+    attemptWriteAndDeleteDynamodbItem(dynamodbDestinationConfig, outputTableName);
+  }
+
+  private static void attemptWriteAndDeleteDynamodbItem(DynamodbDestinationConfig dynamodbDestinationConfig, String outputTableName)
+      throws Exception {
+    DynamoDB dynamoDB = new DynamoDB(getAmazonDynamoDB(dynamodbDestinationConfig));
+    Table table = dynamoDB.createTable(outputTableName, // create table
+        Arrays.asList(new KeySchemaElement(JavaBaseConstants.COLUMN_NAME_AB_ID, KeyType.HASH), new KeySchemaElement("sync_time", KeyType.RANGE)),
+        Arrays.asList(new AttributeDefinition(JavaBaseConstants.COLUMN_NAME_AB_ID, ScalarAttributeType.S),
+            new AttributeDefinition("sync_time", ScalarAttributeType.N)),
+        new ProvisionedThroughput(1L, 1L));
+    table.waitForActive();
+
+    try {
+      PutItemOutcome outcome = table
+          .putItem(
+              new Item().withPrimaryKey(JavaBaseConstants.COLUMN_NAME_AB_ID, UUID.randomUUID().toString(), "sync_time", System.currentTimeMillis()));
+    } catch (Exception e) {
+      LOGGER.error(e.getMessage());
+    }
+
+    table.delete(); // delete table
+    table.waitForDelete();
+  }
+
+  public static AmazonDynamoDB getAmazonDynamoDB(DynamodbDestinationConfig dynamodbDestinationConfig) {
+    var endpoint = dynamodbDestinationConfig.getEndpoint();
+    var region = dynamodbDestinationConfig.getRegion();
+    var accessKeyId = dynamodbDestinationConfig.getAccessKeyId();
+    var secretAccessKey = dynamodbDestinationConfig.getSecretAccessKey();
+
+    var awsCreds = new BasicAWSCredentials(accessKeyId, secretAccessKey);
+
+    if (endpoint.isEmpty()) {
+      return AmazonDynamoDBClientBuilder.standard()
+          .withCredentials(new AWSStaticCredentialsProvider(awsCreds))
+          .withRegion(dynamodbDestinationConfig.getRegion())
+          .build();
+
+    } else {
+      ClientConfiguration clientConfiguration = new ClientConfiguration();
+      clientConfiguration.setSignerOverride("AWSDynamodbSignerType");
+
+      return AmazonDynamoDBClientBuilder
+          .standard()
+          .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint, region))
+          .withClientConfiguration(clientConfiguration)
+          .withCredentials(new AWSStaticCredentialsProvider(awsCreds))
+          .build();
+    }
+  }
+
+}

airbyte-integrations/connectors/destination-dynamodb/src/main/java/io/airbyte/integrations/destination/dynamodb/DynamodbConsumer.java

@@ -0,0 +1,133 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.integrations.destination.dynamodb;
+
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
+import io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer;
+import io.airbyte.protocol.models.*;
+import java.util.*;
+import java.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DynamodbConsumer extends FailureTrackingAirbyteMessageConsumer {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(DynamodbConsumer.class);
+
+  private final DynamodbDestinationConfig dynamodbDestinationConfig;
+  private final ConfiguredAirbyteCatalog configuredCatalog;
+  private final Consumer<AirbyteMessage> outputRecordCollector;
+  private final Map<AirbyteStreamNameNamespacePair, DynamodbWriter> streamNameAndNamespaceToWriters;
+
+  private AirbyteMessage lastStateMessage = null;
+
+  public DynamodbConsumer(DynamodbDestinationConfig dynamodbDestinationConfig,
+                          ConfiguredAirbyteCatalog configuredCatalog,
+                          Consumer<AirbyteMessage> outputRecordCollector) {
+    this.dynamodbDestinationConfig = dynamodbDestinationConfig;
+    this.configuredCatalog = configuredCatalog;
+    this.outputRecordCollector = outputRecordCollector;
+    this.streamNameAndNamespaceToWriters = new HashMap<>(configuredCatalog.getStreams().size());
+  }
+
+  @Override
+  protected void startTracked() throws Exception {
+
+    var endpoint = dynamodbDestinationConfig.getEndpoint();
+    AWSCredentials awsCreds = new BasicAWSCredentials(dynamodbDestinationConfig.getAccessKeyId(), dynamodbDestinationConfig.getSecretAccessKey());
+    AmazonDynamoDB amazonDynamodb = null;
+
+    if (endpoint.isEmpty()) {
+      amazonDynamodb = AmazonDynamoDBClientBuilder.standard()
+          .withCredentials(new AWSStaticCredentialsProvider(awsCreds))
+          .withRegion(dynamodbDestinationConfig.getRegion())
+          .build();
+    } else {
+      ClientConfiguration clientConfiguration = new ClientConfiguration();
+      clientConfiguration.setSignerOverride("AWSDynamodbSignerType");
+
+      amazonDynamodb = AmazonDynamoDBClientBuilder
+          .standard()
+          .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint, dynamodbDestinationConfig.getRegion()))
+          .withClientConfiguration(clientConfiguration)
+          .withCredentials(new AWSStaticCredentialsProvider(awsCreds))
+          .build();
+    }
+
+    var uploadTimestamp = System.currentTimeMillis();
+
+    for (ConfiguredAirbyteStream configuredStream : configuredCatalog.getStreams()) {
+      var writer = new DynamodbWriter(dynamodbDestinationConfig, amazonDynamodb, configuredStream, uploadTimestamp);
+
+      AirbyteStream stream = configuredStream.getStream();
+      AirbyteStreamNameNamespacePair streamNamePair = AirbyteStreamNameNamespacePair
+          .fromAirbyteSteam(stream);
+      streamNameAndNamespaceToWriters.put(streamNamePair, writer);
+    }
+  }
+
+  @Override
+  protected void acceptTracked(AirbyteMessage airbyteMessage) throws Exception {
+    if (airbyteMessage.getType() == AirbyteMessage.Type.STATE) {
+      this.lastStateMessage = airbyteMessage;
+      return;
+    } else if (airbyteMessage.getType() != AirbyteMessage.Type.RECORD) {
+      return;
+    }
+
+    AirbyteRecordMessage recordMessage = airbyteMessage.getRecord();
+    AirbyteStreamNameNamespacePair pair = AirbyteStreamNameNamespacePair
+        .fromRecordMessage(recordMessage);
+
+    if (!streamNameAndNamespaceToWriters.containsKey(pair)) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Message contained record from a stream that was not in the catalog. \ncatalog: %s , \nmessage: %s",
+              Jsons.serialize(configuredCatalog), Jsons.serialize(recordMessage)));
+    }
+
+    streamNameAndNamespaceToWriters.get(pair).write(UUID.randomUUID(), recordMessage);
+  }
+
+  @Override
+  protected void close(boolean hasFailed) throws Exception {
+    for (DynamodbWriter handler : streamNameAndNamespaceToWriters.values()) {
+      handler.close(hasFailed);
+    }
+    // DynamoDB stream uploader is all or nothing if a failure happens in the destination.
+    if (!hasFailed) {
+      outputRecordCollector.accept(lastStateMessage);
+    }
+  }
+
+}