🎉 New Destination: Google Cloud Storage

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

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

airbyte-integrations/connectors/destination-gcs/Dockerfile

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

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

@@ -0,0 +1,39 @@
+plugins {
+    id 'application'
+    id 'airbyte-docker'
+    id 'airbyte-integration-test-java'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.destination.gcs.GcsDestination'
+}
+
+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 project(':airbyte-integrations:connectors:destination-s3')
+    implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
+
+    implementation platform('com.amazonaws:aws-java-sdk-bom:1.12.14')
+    implementation 'com.google.cloud.bigdataoss:gcs-connector:hadoop3-2.2.1'
+    // implementation 'com.amazonaws:aws-java-sdk-s3'
+
+    // csv
+    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'
+
+    // parquet
+    implementation group: 'org.apache.hadoop', name: 'hadoop-common', version: '3.3.0'
+    implementation group: 'org.apache.hadoop', name: 'hadoop-aws', version: '3.3.0'
+    implementation group: 'org.apache.hadoop', name: 'hadoop-mapreduce-client-core', version: '3.3.0'
+    implementation group: 'org.apache.parquet', name: 'parquet-avro', version: '1.12.0'
+    implementation group: 'tech.allegro.schema.json2avro', name: 'converter', version: '0.2.10'
+
+    testImplementation 'org.apache.commons:commons-lang3:3.11'
+
+    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
+    integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-gcs')
+}

airbyte-integrations/connectors/destination-gcs/readme.md

@@ -0,0 +1,6 @@
+# Destination Google Cloud Storage (GCS)
+
+In order to test the GCS destination, you need a Google Cloud Platform account.
+
+You need to generate the HMAC Keys for a service account then use them as keys to connect.
+Obviously you also need to provide the right grants to the user in order to allow it to create and delete buckets and files in GCS.

airbyte-integrations/connectors/destination-gcs/src/main/java/io/airbyte/integrations/destination/gcs/GcsConsumer.java

@@ -0,0 +1,119 @@
+/*
+ * 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.gcs;
+
+import com.amazonaws.services.s3.AmazonS3;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
+import io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer;
+import io.airbyte.integrations.destination.gcs.writer.GcsWriterFactory;
+import io.airbyte.integrations.destination.s3.writer.S3Writer;
+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;
+import java.util.function.Consumer;
+
+public class GcsConsumer extends FailureTrackingAirbyteMessageConsumer {
+
+  private final GcsDestinationConfig gcsDestinationConfig;
+  private final ConfiguredAirbyteCatalog configuredCatalog;
+  private final GcsWriterFactory writerFactory;
+  private final Consumer<AirbyteMessage> outputRecordCollector;
+  private final Map<AirbyteStreamNameNamespacePair, S3Writer> streamNameAndNamespaceToWriters;
+
+  private AirbyteMessage lastStateMessage = null;
+
+  public GcsConsumer(GcsDestinationConfig gcsDestinationConfig,
+                     ConfiguredAirbyteCatalog configuredCatalog,
+                     GcsWriterFactory writerFactory,
+                     Consumer<AirbyteMessage> outputRecordCollector) {
+    this.gcsDestinationConfig = gcsDestinationConfig;
+    this.configuredCatalog = configuredCatalog;
+    this.writerFactory = writerFactory;
+    this.outputRecordCollector = outputRecordCollector;
+    this.streamNameAndNamespaceToWriters = new HashMap<>(configuredCatalog.getStreams().size());
+  }
+
+  @Override
+  protected void startTracked() throws Exception {
+    AmazonS3 s3Client = GcsS3Helper.getGcsS3Client(gcsDestinationConfig);
+
+    Timestamp uploadTimestamp = new Timestamp(System.currentTimeMillis());
+
+    for (ConfiguredAirbyteStream configuredStream : configuredCatalog.getStreams()) {
+      S3Writer writer = writerFactory
+          .create(gcsDestinationConfig, s3Client, configuredStream, uploadTimestamp);
+      writer.initialize();
+
+      AirbyteStream stream = configuredStream.getStream();
+      AirbyteStreamNameNamespacePair streamNamePair = AirbyteStreamNameNamespacePair
+          .fromAirbyteSteam(stream);
+      streamNameAndNamespaceToWriters.put(streamNamePair, writer);
+    }
+  }
+
+  @Override
+  protected void acceptTracked(AirbyteMessage airbyteMessage) throws Exception {
+    if (airbyteMessage.getType() == Type.STATE) {
+      this.lastStateMessage = airbyteMessage;
+      return;
+    } else if (airbyteMessage.getType() != 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)));
+    }
+
+    UUID id = UUID.randomUUID();
+    streamNameAndNamespaceToWriters.get(pair).write(id, recordMessage);
+  }
+
+  @Override
+  protected void close(boolean hasFailed) throws Exception {
+    for (S3Writer handler : streamNameAndNamespaceToWriters.values()) {
+      handler.close(hasFailed);
+    }
+    // Gcs stream uploader is all or nothing if a failure happens in the destination.
+    if (!hasFailed) {
+      outputRecordCollector.accept(lastStateMessage);
+    }
+  }
+
+}

airbyte-integrations/connectors/destination-gcs/src/main/java/io/airbyte/integrations/destination/gcs/GcsDestination.java

@@ -0,0 +1,90 @@
+/*
+ * 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.gcs;
+
+import com.amazonaws.services.s3.AmazonS3;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import io.airbyte.integrations.BaseConnector;
+import io.airbyte.integrations.base.AirbyteMessageConsumer;
+import io.airbyte.integrations.base.Destination;
+import io.airbyte.integrations.base.IntegrationRunner;
+import io.airbyte.integrations.destination.gcs.writer.GcsWriterFactory;
+import io.airbyte.integrations.destination.gcs.writer.ProductionWriterFactory;
+import io.airbyte.integrations.destination.jdbc.copy.gcs.GcsConfig;
+import io.airbyte.integrations.destination.jdbc.copy.gcs.GcsStreamCopier;
+import io.airbyte.protocol.models.AirbyteConnectionStatus;
+import io.airbyte.protocol.models.AirbyteConnectionStatus.Status;
+import io.airbyte.protocol.models.AirbyteMessage;
+import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import java.io.IOException;
+import java.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GcsDestination extends BaseConnector implements Destination {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(GcsDestination.class);
+
+  public static void main(String[] args) throws Exception {
+    new IntegrationRunner(new GcsDestination()).run(args);
+  }
+
+  @Override
+  public AirbyteConnectionStatus check(JsonNode config) {
+    try {
+      GcsDestinationConfig destinationConfig = GcsDestinationConfig.getGcsDestinationConfig(config);
+      AmazonS3 s3Client = GcsS3Helper.getGcsS3Client(destinationConfig);
+      s3Client.putObject(destinationConfig.getBucketName(), "test", "check-content");
+      s3Client.deleteObject(destinationConfig.getBucketName(), "test");
+      return new AirbyteConnectionStatus().withStatus(Status.SUCCEEDED);
+    } catch (Exception e) {
+      LOGGER.error("Exception attempting to access the Gcs bucket: {}", e.getMessage());
+      return new AirbyteConnectionStatus()
+          .withStatus(AirbyteConnectionStatus.Status.FAILED)
+          .withMessage("Could not connect to the Gcs bucket with the provided configuration. \n" + e
+              .getMessage());
+    }
+  }
+
+  @Override
+  public AirbyteMessageConsumer getConsumer(JsonNode config,
+                                            ConfiguredAirbyteCatalog configuredCatalog,
+                                            Consumer<AirbyteMessage> outputRecordCollector) {
+    GcsWriterFactory formatterFactory = new ProductionWriterFactory();
+    return new GcsConsumer(GcsDestinationConfig.getGcsDestinationConfig(config), configuredCatalog, formatterFactory, outputRecordCollector);
+  }
+
+  private static void attemptWriteAndDeleteGcsObject(GcsConfig gcsConfig) throws IOException {
+    var storage = GcsStreamCopier.getStorageClient(gcsConfig);
+    var blobId = BlobId.of(gcsConfig.getBucketName(), "check-content/test-file");
+    var blobInfo = BlobInfo.newBuilder(blobId).build();
+
+    storage.create(blobInfo, "".getBytes());
+    storage.delete(blobId);
+  }
+
+}

airbyte-integrations/connectors/destination-gcs/src/main/java/io/airbyte/integrations/destination/gcs/GcsDestinationConfig.java

@@ -0,0 +1,82 @@
+/*
+ * 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.gcs;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.integrations.destination.gcs.credential.GcsCredentialConfig;
+import io.airbyte.integrations.destination.gcs.credential.GcsCredentialConfigs;
+import io.airbyte.integrations.destination.s3.S3FormatConfig;
+import io.airbyte.integrations.destination.s3.S3FormatConfigs;
+
+public class GcsDestinationConfig {
+
+  private final String bucketName;
+  private final String bucketPath;
+  private final String bucketRegion;
+  private final GcsCredentialConfig credentialConfig;
+  private final S3FormatConfig formatConfig;
+
+  public GcsDestinationConfig(String bucketName,
+                              String bucketPath,
+                              String bucketRegion,
+                              GcsCredentialConfig credentialConfig,
+                              S3FormatConfig formatConfig) {
+    this.bucketName = bucketName;
+    this.bucketPath = bucketPath;
+    this.bucketRegion = bucketRegion;
+    this.credentialConfig = credentialConfig;
+    this.formatConfig = formatConfig;
+  }
+
+  public static GcsDestinationConfig getGcsDestinationConfig(JsonNode config) {
+    return new GcsDestinationConfig(
+        config.get("gcs_bucket_name").asText(),
+        config.get("gcs_bucket_path").asText(),
+        config.get("gcs_bucket_region").asText(),
+        GcsCredentialConfigs.getCredentialConfig(config),
+        S3FormatConfigs.getS3FormatConfig(config));
+  }
+
+  public String getBucketName() {
+    return bucketName;
+  }
+
+  public String getBucketPath() {
+    return bucketPath;
+  }
+
+  public String getBucketRegion() {
+    return bucketRegion;
+  }
+
+  public GcsCredentialConfig getCredentialConfig() {
+    return credentialConfig;
+  }
+
+  public S3FormatConfig getFormatConfig() {
+    return formatConfig;
+  }
+
+}