Implement destination null

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

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

airbyte-integrations/connectors/destination-null/Dockerfile

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

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

@@ -0,0 +1,68 @@
+# Destination Null
+
+This is the repository for the Null destination connector in Java.
+For information about how to use this connector within Airbyte, see [the User Documentation](https://docs.airbyte.io/integrations/destinations/null).
+
+## Local development
+
+#### Building via Gradle
+From the Airbyte repository root, run:
+```
+./gradlew :airbyte-integrations:connectors:destination-null: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-null: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-null:dev spec
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/destination-null:dev check --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/destination-null:dev discover --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets -v $(pwd)/integration_tests:/integration_tests airbyte/destination-null: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/null`.
+
+#### 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/nullDestinationAcceptanceTest.java`.
+
+### Using gradle to run tests
+All commands should be run from airbyte project root.
+To run unit tests:
+```
+./gradlew :airbyte-integrations:connectors:destination-null:unitTest
+```
+To run acceptance and custom integration tests:
+```
+./gradlew :airbyte-integrations:connectors:destination-null: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-null/build.gradle

@@ -0,0 +1,19 @@
+plugins {
+    id 'application'
+    id 'airbyte-docker'
+    id 'airbyte-integration-test-java'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.destination.destination_null.NullDestination'
+}
+
+dependencies {
+    implementation project(':airbyte-config:models')
+    implementation project(':airbyte-protocol:models')
+    implementation project(':airbyte-integrations:bases:base-java')
+    implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
+
+    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
+    integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-null')
+}

airbyte-integrations/connectors/destination-null/src/main/java/io/airbyte/integrations/destination/destination_null/NullConsumer.java

@@ -0,0 +1,76 @@
+package io.airbyte.integrations.destination.destination_null;
+
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
+import io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer;
+import io.airbyte.integrations.destination.destination_null.logger.NullDestinationLogger;
+import io.airbyte.integrations.destination.destination_null.logger.NullDestinationLoggerFactory;
+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.util.HashMap;
+import java.util.Map;
+import java.util.function.Consumer;
+
+public class NullConsumer extends FailureTrackingAirbyteMessageConsumer {
+
+  private final NullDestinationLoggerFactory loggerFactory;
+  private final ConfiguredAirbyteCatalog configuredCatalog;
+  private final Consumer<AirbyteMessage> outputRecordCollector;
+  private final Map<AirbyteStreamNameNamespacePair, NullDestinationLogger> loggers;
+
+  private AirbyteMessage lastStateMessage = null;
+
+  public NullConsumer(NullDestinationLoggerFactory loggerFactory,
+                      ConfiguredAirbyteCatalog configuredCatalog,
+                      Consumer<AirbyteMessage> outputRecordCollector) {
+    this.loggerFactory = loggerFactory;
+    this.configuredCatalog = configuredCatalog;
+    this.outputRecordCollector = outputRecordCollector;
+    this.loggers = new HashMap<>();
+  }
+
+  @Override
+  protected void startTracked() {
+    for (ConfiguredAirbyteStream configuredStream : configuredCatalog.getStreams()) {
+      final AirbyteStream stream = configuredStream.getStream();
+      final AirbyteStreamNameNamespacePair streamNamePair = AirbyteStreamNameNamespacePair.fromAirbyteSteam(stream);
+      final NullDestinationLogger logger = loggerFactory.create(streamNamePair);
+      loggers.put(streamNamePair, logger);
+    }
+  }
+
+  @Override
+  protected void acceptTracked(AirbyteMessage airbyteMessage) {
+    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 (!loggers.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)));
+    }
+
+    loggers.get(pair).log(recordMessage);
+  }
+
+  @Override
+  protected void close(boolean hasFailed) {
+    if (!hasFailed) {
+      outputRecordCollector.accept(lastStateMessage);
+    }
+  }
+
+}

airbyte-integrations/connectors/destination-null/src/main/java/io/airbyte/integrations/destination/destination_null/NullDestination.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.destination_null;
+
+import com.fasterxml.jackson.databind.JsonNode;
+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.destination_null.logger.NullDestinationLogger;
+import io.airbyte.integrations.destination.destination_null.logger.NullDestinationLoggerFactory;
+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.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NullDestination extends BaseConnector implements Destination {
+
+  public static void main(String[] args) throws Exception {
+    new IntegrationRunner(new NullDestination()).run(args);
+  }
+
+  /**
+   * The null destination is always available!
+   */
+  @Override
+  public AirbyteConnectionStatus check(JsonNode config) {
+    return new AirbyteConnectionStatus().withStatus(Status.SUCCEEDED);
+  }
+
+  @Override
+  public AirbyteMessageConsumer getConsumer(JsonNode config,
+                                            ConfiguredAirbyteCatalog configuredCatalog,
+                                            Consumer<AirbyteMessage> outputRecordCollector) {
+    return new NullConsumer(new NullDestinationLoggerFactory(config), configuredCatalog, outputRecordCollector);
+  }
+
+}

airbyte-integrations/connectors/destination-null/src/main/java/io/airbyte/integrations/destination/destination_null/logger/BaseLogger.java

@@ -0,0 +1,45 @@
+package io.airbyte.integrations.destination.destination_null.logger;
+
+import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
+import io.airbyte.protocol.models.AirbyteRecordMessage;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class BaseLogger implements NullDestinationLogger {
+
+  protected final AirbyteStreamNameNamespacePair streamNamePair;
+  protected final int maxEntryCount;
+  protected int loggedEntryCount = 0;
+
+  public BaseLogger(AirbyteStreamNameNamespacePair streamNamePair, int maxEntryCount) {
+    this.streamNamePair = streamNamePair;
+    this.maxEntryCount = maxEntryCount;
+  }
+
+  protected String entryMessage(AirbyteRecordMessage recordMessage) {
+    return String.format("[%s] %s #%04d: %s",
+        emissionTimestamp(recordMessage.getEmittedAt()),
+        streamName(streamNamePair),
+        loggedEntryCount,
+        recordMessage.getData());
+  }
+
+  protected static String streamName(AirbyteStreamNameNamespacePair pair) {
+    if (pair.getNamespace() == null) {
+      return pair.getName();
+    } else {
+      return String.format("%s.%s", pair.getNamespace(), pair.getName());
+    }
+  }
+
+  protected static String emissionTimestamp(long emittedAt) {
+    return OffsetDateTime
+        .ofInstant(Instant.ofEpochMilli(emittedAt), ZoneId.systemDefault())
+        .format(DateTimeFormatter.ISO_OFFSET_DATE_TIME);
+  }
+
+}

airbyte-integrations/connectors/destination-null/src/main/java/io/airbyte/integrations/destination/destination_null/logger/DevNull.java

@@ -0,0 +1,17 @@
+package io.airbyte.integrations.destination.destination_null.logger;
+
+import io.airbyte.protocol.models.AirbyteRecordMessage;
+
+public class DevNull implements NullDestinationLogger {
+
+  public static final DevNull SINGLETON = new DevNull();
+
+  private DevNull() {
+  }
+
+  @Override
+  public void log(AirbyteRecordMessage recordMessage) {
+    // nothing happens in /dev/null
+  }
+
+}

airbyte-integrations/connectors/destination-null/src/main/java/io/airbyte/integrations/destination/destination_null/logger/EveryNthLogger.java

@@ -0,0 +1,33 @@
+package io.airbyte.integrations.destination.destination_null.logger;
+
+import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
+import io.airbyte.protocol.models.AirbyteRecordMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class EveryNthLogger extends BaseLogger implements NullDestinationLogger {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(EveryNthLogger.class);
+
+  private final int nthEntryToLog;
+  private int currentEntry = 0;
+
+  public EveryNthLogger(AirbyteStreamNameNamespacePair streamNamePair, int nthEntryToLog, int maxEntryCount) {
+    super(streamNamePair, maxEntryCount);
+    this.nthEntryToLog = nthEntryToLog;
+  }
+
+  @Override
+  public void log(AirbyteRecordMessage recordMessage) {
+    if (loggedEntryCount >= maxEntryCount) {
+      return;
+    }
+
+    currentEntry += 1;
+    if (currentEntry % nthEntryToLog == 0) {
+      loggedEntryCount += 1;
+      LOGGER.info(entryMessage(recordMessage));
+    }
+  }
+
+}

airbyte-integrations/connectors/destination-null/src/main/java/io/airbyte/integrations/destination/destination_null/logger/FirstNLogger.java

@@ -0,0 +1,26 @@
+package io.airbyte.integrations.destination.destination_null.logger;
+
+import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
+import io.airbyte.protocol.models.AirbyteRecordMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FirstNLogger extends BaseLogger implements NullDestinationLogger {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(FirstNLogger.class);
+
+  public FirstNLogger(AirbyteStreamNameNamespacePair streamNamePair, int maxEntryCount) {
+    super(streamNamePair, maxEntryCount);
+  }
+
+  @Override
+  public void log(AirbyteRecordMessage recordMessage) {
+    if (loggedEntryCount >= maxEntryCount) {
+      return;
+    }
+
+    loggedEntryCount += 1;
+    LOGGER.info(entryMessage(recordMessage));
+  }
+
+}

airbyte-integrations/connectors/destination-null/src/main/java/io/airbyte/integrations/destination/destination_null/logger/NullDestinationLogger.java

@@ -0,0 +1,16 @@
+package io.airbyte.integrations.destination.destination_null.logger;
+
+import io.airbyte.protocol.models.AirbyteRecordMessage;
+
+public interface NullDestinationLogger {
+
+  enum LoggingType {
+    NoLogging,
+    FirstN,
+    EveryNth,
+    RandomSampling
+  }
+
+  void log(AirbyteRecordMessage recordMessage);
+
+}