🎉 New destination: S3

README.md

@@ -1,6 +1,6 @@
 # Introduction
 
-![GitHub Workflow Status](https://img.shields.io/github/workflow/status/airbytehq/airbyte/Airbyte%20CI) ![License](https://img.shields.io/github/license/airbytehq/airbyte)
+[![GitHub Workflow Status](https://img.shields.io/github/workflow/status/airbytehq/airbyte/Airbyte%20CI)](https://github.com/airbytehq/airbyte/actions/workflows/gradle.yml) [![License](https://img.shields.io/github/license/airbytehq/airbyte)](./LICENSE)
 
 ![](docs/.gitbook/assets/airbyte_horizontal_color_white-background.svg)
 

airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/7679c5ea-e829-4202-bd47-61ae9debdds3.json

@@ -0,0 +1,7 @@
+{
+  "destinationDefinitionId": "7679c5ea-e829-4202-bd47-61ae9debdds3",
+  "name": "S3",
+  "dockerRepository": "airbyte/destination-s3",
+  "dockerImageTag": "0.1.0",
+  "documentationUrl": "https://docs.airbyte.io/integrations/destinations/s3"
+}

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

@@ -24,6 +24,11 @@
   dockerRepository: airbyte/destination-snowflake
   dockerImageTag: 0.3.6
   documentationUrl: https://docs.airbyte.io/integrations/destinations/snowflake
+- destinationDefinitionId: 7679c5ea-e829-4202-bd47-61ae9debdds3
+  name: S3
+  dockerRepository: airbyte/destination-s3
+  dockerImageTag: 0.1.0
+  documentationUrl: https://docs.airbyte.io/integrations/destinations/s3
 - destinationDefinitionId: f7a7d195-377f-cf5b-70a5-be6b819019dc
   name: Redshift
   dockerRepository: airbyte/destination-redshift

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/FailureTrackingAirbyteMessageConsumer.java

@@ -74,7 +74,11 @@ public void accept(AirbyteMessage msg) throws Exception {
 
   @Override
   public void close() throws Exception {
-    LOGGER.info("hasFailed: {}.", hasFailed);
+    if (hasFailed) {
+      LOGGER.warn("hasFailed: true.");
+    } else {
+      LOGGER.info("hasFailed: false.");
+    }
     close(hasFailed);
   }
 

airbyte-integrations/connectors/destination-bigquery/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryDestination.java

@@ -52,7 +52,7 @@
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.lang.Exceptions;
-import io.airbyte.commons.resources.MoreResources;
+import io.airbyte.integrations.BaseConnector;
 import io.airbyte.integrations.base.AirbyteMessageConsumer;
 import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
 import io.airbyte.integrations.base.Destination;
@@ -68,7 +68,6 @@
 import io.airbyte.protocol.models.AirbyteStream;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import io.airbyte.protocol.models.ConfiguredAirbyteStream;
-import io.airbyte.protocol.models.ConnectorSpecification;
 import io.airbyte.protocol.models.DestinationSyncMode;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
@@ -84,7 +83,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class BigQueryDestination implements Destination {
+public class BigQueryDestination extends BaseConnector implements Destination {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryDestination.class);
   static final String CONFIG_DATASET_ID = "dataset_id";
@@ -102,13 +101,6 @@ public BigQueryDestination() {
     namingResolver = new StandardNameTransformer();
   }
 
-  @Override
-  public ConnectorSpecification spec() throws IOException {
-    // return a jsonschema representation of the spec for the integration.
-    final String resourceString = MoreResources.readResource("spec.json");
-    return Jsons.deserialize(resourceString, ConnectorSpecification.class);
-  }
-
   @Override
   public AirbyteConnectionStatus check(JsonNode config) {
     try {

airbyte-integrations/connectors/destination-csv/src/main/java/io/airbyte/integrations/destination/csv/CsvDestination.java

@@ -27,7 +27,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.base.Preconditions;
 import io.airbyte.commons.json.Jsons;
-import io.airbyte.commons.resources.MoreResources;
+import io.airbyte.integrations.BaseConnector;
 import io.airbyte.integrations.base.AirbyteMessageConsumer;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer;
@@ -41,7 +41,6 @@
 import io.airbyte.protocol.models.AirbyteRecordMessage;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import io.airbyte.protocol.models.ConfiguredAirbyteStream;
-import io.airbyte.protocol.models.ConnectorSpecification;
 import io.airbyte.protocol.models.DestinationSyncMode;
 import java.io.FileWriter;
 import java.io.IOException;
@@ -59,7 +58,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class CsvDestination implements Destination {
+public class CsvDestination extends BaseConnector implements Destination {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(CsvDestination.class);
 
@@ -71,12 +70,6 @@ public CsvDestination() {
     namingResolver = new StandardNameTransformer();
   }
 
-  @Override
-  public ConnectorSpecification spec() throws IOException {
-    final String resourceString = MoreResources.readResource("spec.json");
-    return Jsons.deserialize(resourceString, ConnectorSpecification.class);
-  }
-
   @Override
   public AirbyteConnectionStatus check(JsonNode config) {
     try {

airbyte-integrations/connectors/destination-local-json/src/main/java/io/airbyte/integrations/destination/local_json/LocalJsonDestination.java

@@ -28,7 +28,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.json.Jsons;
-import io.airbyte.commons.resources.MoreResources;
+import io.airbyte.integrations.BaseConnector;
 import io.airbyte.integrations.base.AirbyteMessageConsumer;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer;
@@ -42,7 +42,6 @@
 import io.airbyte.protocol.models.AirbyteRecordMessage;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import io.airbyte.protocol.models.ConfiguredAirbyteStream;
-import io.airbyte.protocol.models.ConnectorSpecification;
 import io.airbyte.protocol.models.DestinationSyncMode;
 import java.io.FileWriter;
 import java.io.IOException;
@@ -59,7 +58,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class LocalJsonDestination implements Destination {
+public class LocalJsonDestination extends BaseConnector implements Destination {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(LocalJsonDestination.class);
 
@@ -71,12 +70,6 @@ public LocalJsonDestination() {
     namingResolver = new StandardNameTransformer();
   }
 
-  @Override
-  public ConnectorSpecification spec() throws IOException {
-    final String resourceString = MoreResources.readResource("spec.json");
-    return Jsons.deserialize(resourceString, ConnectorSpecification.class);
-  }
-
   @Override
   public AirbyteConnectionStatus check(JsonNode config) {
     try {

airbyte-integrations/connectors/destination-redshift/src/main/resources/spec.json

@@ -78,7 +78,6 @@
           "ap-northeast-3",
           "ap-southeast-1",
           "ap-southeast-2",
-          "ap-northeast-1",
           "ca-central-1",
           "cn-north-1",
           "cn-northwest-1",

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

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

airbyte-integrations/connectors/destination-s3/Dockerfile

@@ -0,0 +1,11 @@
+FROM airbyte/integration-base-java:dev
+
+WORKDIR /airbyte
+ENV APPLICATION destination-s3
+
+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-s3

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

@@ -0,0 +1,23 @@
+plugins {
+    id 'application'
+    id 'airbyte-docker'
+    id 'airbyte-integration-test-java'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.destination.s3.S3Destination'
+}
+
+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-s3:1.11.978'
+    implementation 'org.apache.commons:commons-csv:1.4'
+    implementation 'com.github.alexmojaki:s3-stream-upload:2.2.2'
+
+    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
+}

airbyte-integrations/connectors/destination-s3/src/main/java/io/airbyte/integrations/destination/s3/S3Consumer.java

@@ -0,0 +1,108 @@
+/*
+ * 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.s3;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
+import io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer;
+import io.airbyte.protocol.models.AirbyteMessage;
+import io.airbyte.protocol.models.AirbyteMessage.Type;
+import io.airbyte.protocol.models.AirbyteRecordMessage;
+import io.airbyte.protocol.models.AirbyteStream;
+import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import io.airbyte.protocol.models.ConfiguredAirbyteStream;
+import java.sql.Timestamp;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+public class S3Consumer extends FailureTrackingAirbyteMessageConsumer {
+
+  private final S3DestinationConfig s3DestinationConfig;
+  private final ConfiguredAirbyteCatalog catalog;
+  private final Map<AirbyteStreamNameNamespacePair, S3Handler> pairToHandlers;
+
+  public S3Consumer(
+                    S3DestinationConfig s3DestinationConfig,
+                    ConfiguredAirbyteCatalog catalog) {
+    this.s3DestinationConfig = s3DestinationConfig;
+    this.catalog = catalog;
+    this.pairToHandlers = new HashMap<>(catalog.getStreams().size());
+  }
+
+  @Override
+  protected void startTracked() throws Exception {
+    AWSCredentials awsCreds = new BasicAWSCredentials(s3DestinationConfig.getAccessKeyId(),
+        s3DestinationConfig.getSecretAccessKey());
+    AmazonS3 s3Client = AmazonS3ClientBuilder.standard()
+        .withCredentials(new AWSStaticCredentialsProvider(awsCreds))
+        .withRegion(s3DestinationConfig.getBucketRegion())
+        .build();
+    Timestamp uploadTimestamp = new Timestamp(System.currentTimeMillis());
+
+    for (ConfiguredAirbyteStream configuredStream : catalog.getStreams()) {
+      S3Handler handler = S3Handlers.getS3Handler(s3DestinationConfig, s3Client, configuredStream, uploadTimestamp);
+      handler.initialize();
+
+      AirbyteStream stream = configuredStream.getStream();
+      AirbyteStreamNameNamespacePair pair = AirbyteStreamNameNamespacePair.fromAirbyteSteam(stream);
+      pairToHandlers.put(pair, handler);
+    }
+  }
+
+  @Override
+  protected void acceptTracked(AirbyteMessage airbyteMessage) throws Exception {
+    if (airbyteMessage.getType() != Type.RECORD) {
+      return;
+    }
+
+    AirbyteRecordMessage recordMessage = airbyteMessage.getRecord();
+    AirbyteStreamNameNamespacePair pair = AirbyteStreamNameNamespacePair
+        .fromRecordMessage(recordMessage);
+
+    if (!pairToHandlers.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(catalog), Jsons.serialize(recordMessage)));
+    }
+
+    UUID id = UUID.randomUUID();
+    pairToHandlers.get(pair).write(id, recordMessage);
+  }
+
+  @Override
+  protected void close(boolean hasFailed) throws Exception {
+    for (S3Handler handler : pairToHandlers.values()) {
+      handler.close(hasFailed);
+    }
+  }
+
+}