java csv destination (#505)

airbyte-integrations/base-java/Dockerfile

@@ -10,6 +10,6 @@ ENV AIRBYTE_SPEC_CMD "./javabase.sh --spec"
 ENV AIRBYTE_CHECK_CMD "./javabase.sh --check"
 ENV AIRBYTE_DISCOVER_CMD "./javabase.sh --discover"
 ENV AIRBYTE_READ_CMD "./javabase.sh --read"
-ENV AIRBYTE_READ_CMD "./javabase.sh --write"
+ENV AIRBYTE_WRITE_CMD "./javabase.sh --write"
 
 ENTRYPOINT ["/airbyte/base.sh"]

airbyte-integrations/base-java/javabase.sh

@@ -2,6 +2,7 @@
 
 set -e
 
-# wrap run script in a script so that we can lazy evaluate the value of APPLICATION. APPLICATION is
+# Wrap run script in a script so that we can lazy evaluate the value of APPLICATION. APPLICATION is
 # set by the dockerfile that inherits base-java, so it cannot be evaluated when base-java is built.
-bin/"$APPLICATION" "$@"
+# We also need to make sure that stdin of the script is piped to the stdin of the java application.
+cat <&0 | bin/"$APPLICATION" "$@"

airbyte-integrations/base-java/src/main/java/io/airbyte/integrations/base/FailureTrackingConsumer.java

@@ -0,0 +1,54 @@
+/*
+ * 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.base;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class FailureTrackingConsumer<T> implements DestinationConsumer<T> {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(FailureTrackingConsumer.class);
+
+  private boolean hasFailed = false;
+
+  protected abstract void acceptTracked(T t) throws Exception;
+
+  public void accept(T t) throws Exception {
+    try {
+      acceptTracked(t);
+    } catch (Exception e) {
+      hasFailed = true;
+      throw e;
+    }
+  }
+
+  protected abstract void close(boolean hasFailed) throws Exception;
+
+  public void close() throws Exception {
+    LOGGER.info("hasFailed: {}.", hasFailed);
+    close(hasFailed);
+  }
+
+}

airbyte-integrations/base-java/src/main/java/io/airbyte/integrations/base/JavaBaseConstants.java

@@ -24,8 +24,6 @@
 
 package io.airbyte.integrations.base;
 
-import java.nio.file.Path;
-
 public class JavaBaseConstants {
 
   public static String ARGS_CONFIG_KEY = "config";
@@ -36,8 +34,4 @@ public class JavaBaseConstants {
   public static String ARGS_CATALOG_DESC = "input path for the catalog";
   public static String ARGS_PATH_DESC = "path to the json-encoded state file";
 
-  // todo (cgardens) - this mount path should be passed in by the worker and read as an arg or
-  // environment variable by the runner.
-  public static Path LOCAL_MOUNT = Path.of("/local");
-
 }

airbyte-integrations/base-java/src/test/java/io/airbyte/integrations/base/FailureTrackingConsumerTest.java

@@ -0,0 +1,71 @@
+/*
+ * 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.base;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+import org.junit.jupiter.api.Test;
+
+class FailureTrackingConsumerTest {
+
+  @Test
+  void testNoFailure() throws Exception {
+    final TestConsumer consumer = spy(new TestConsumer());
+    consumer.accept("");
+    consumer.close();
+
+    verify(consumer).close(false);
+  }
+
+  @Test
+  void testWithFailure() throws Exception {
+    final TestConsumer consumer = spy(new TestConsumer());
+    doThrow(new RuntimeException()).when(consumer).acceptTracked("");
+
+    // verify the exception still gets thrown.
+    assertThrows(RuntimeException.class, () -> consumer.accept(""));
+    consumer.close();
+
+    verify(consumer).close(true);
+  }
+
+  static class TestConsumer extends FailureTrackingConsumer<String> {
+
+    @Override
+    protected void acceptTracked(String s) {
+
+    }
+
+    @Override
+    protected void close(boolean hasFailed) {
+
+    }
+
+  }
+
+}

airbyte-integrations/base/base.sh

@@ -51,6 +51,9 @@ function main() {
     # todo: state should be optional: --state "$STATE_FILE"
     eval "$AIRBYTE_READ_CMD" --config "$CONFIG_FILE" --catalog "$CATALOG_FILE"
     ;;
+  write)
+    eval "$AIRBYTE_WRITE_CMD" --config "$CONFIG_FILE" --catalog "$CATALOG_FILE"
+    ;;
   *)
     error "Unknown command: $CMD"
     ;;

airbyte-integrations/csv-destination/.dockerignore

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

airbyte-integrations/csv-destination/Dockerfile

@@ -0,0 +1,8 @@
+FROM airbyte/base-java:dev
+
+WORKDIR /airbyte
+ENV APPLICATION csv-destination
+
+COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
+
+RUN tar xf ${APPLICATION}.tar --strip-components=1

airbyte-integrations/csv-destination/build.gradle

@@ -0,0 +1,31 @@
+import com.bmuschko.gradle.docker.tasks.image.DockerBuildImage
+plugins {
+    id 'com.bmuschko.docker-remote-api'
+    id 'application'
+}
+dependencies {
+    implementation project(':airbyte-config:models')
+    implementation project(':airbyte-singer')
+    implementation project(':airbyte-integrations:base-java')
+
+    implementation 'org.apache.commons:commons-csv:1.4'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.destination.csv.CsvDestination'
+}
+
+
+def image = 'airbyte/airbyte-csv-destination:dev'
+
+task imageName {
+    doLast {
+        println "IMAGE $image"
+    }
+}
+
+task buildImage(type: DockerBuildImage) {
+    inputDir = projectDir
+    images.add(image)
+    dependsOn ':airbyte-integrations:base-java:buildImage'
+}