Integration Test and Deadletter queue support (#23)

* 1.Deadletter topic support to send data to deadtopic in case AEP connector fails to send data to AEP after trying n retries.
2. Integration test for verifying common scenario to check if main use case is not effected due to changes

* Fix java 8 issue to support backward compatible

Co-authored-by: vivetiwa <[email protected]>

Author: vivetiwa

DEVELOPER_GUIDE.md

@@ -27,19 +27,21 @@ To install the connector, drop the jar file into the plug in directory of Kafka
 AEP Sink connector configurations can be supplied in the call register the connector.
 
 
-| Config Name                       | Description                                   | Default                                                 | Required | Example |
-|-----------------------------------|-----------------------------------------------|---------------------------------------------------------|----------|---------|
-| topics                            | comma separated list of topics                |                                                         | yes      |         |
-| connector.class                   | classname of impl                             | com.adobe.platform.streaming.sink.impl.AEPSinkConnector | yes      |         |
-| key.converter.schemas.enable      | enables conversion of schemas                 | false                                                   | no       |         |
-| value.converter.schemas.enable    | enables conversion of schemas                 | false                                                   | no       |         |
-| aep.endpoint                      | aep streaming endpoint url                    |                                                         | yes      |         |
-| aep.connection.auth.enabled       | required for authenticated streaming endpoint | false                                                   | no       |         |
-| aep.connection.auth.token.type    | always set to access_token                    | access_token                                            | no       |         |
-| aep.connection.auth.client.id     | IMS client id                                 |                                                         | no       |         |
-| aep.connection.auth.client.code   | IMS client code                               |                                                         | no       |         |
-| aep.connection.auth.client.secret | IME client secret                             |                                                         |          |         |
-| aep.flush.bytes.kb                | bytes threshold to determine the batch        | 4                                                       | no       |         |
+| Config Name                       | Description                                     | Default                                                 | Required | Example                 |
+|-----------------------------------|-------------------------------------------------|---------------------------------------------------------|----------|-------------------------|
+| topics                            | comma separated list of topics                  |                                                         | yes      |                         |
+| connector.class                   | classname of impl                               | com.adobe.platform.streaming.sink.impl.AEPSinkConnector | yes      |                         |
+| key.converter.schemas.enable      | enables conversion of schemas                   | false                                                   | no       |                         |
+| value.converter.schemas.enable    | enables conversion of schemas                   | false                                                   | no       |                         |
+| aep.endpoint                      | aep streaming endpoint url                      |                                                         | yes      |                         |
+| aep.connection.auth.enabled       | required for authenticated streaming endpoint   | false                                                   | no       |                         |
+| aep.connection.auth.token.type    | always set to access_token                      | access_token                                            | no       |                         |
+| aep.connection.auth.client.id     | IMS client id                                   |                                                         | no       |                         |
+| aep.connection.auth.client.code   | IMS client code                                 |                                                         | no       |                         |
+| aep.connection.auth.client.secret | IME client secret                               |                                                         | no       |                         |
+| aep.flush.bytes.kb                | bytes threshold to determine the batch          | 4                                                       | no       |                         |
+| aep.error.logger                  | put failed message to dead letter topic or log  | none                                                    | no       |  kafka, log, both, none |
+| aep.error.topic                   | deadletter topic name                           | none                                                    | no       |                         |
 
 ## Step-by-Step Workflow
 

dependencies.gradle

@@ -29,6 +29,7 @@ versions.orgJson = '20180130'
 versions.kafkaConnect = '2.8.0'
 versions.slf4j = '1.7.25'
 versions.jaxb = '2.3.0'
+versions.wiremock = '2.27.2'
 
 libraries.collections4 = [
   "org.apache.commons:commons-collections4:$versions.collections4"
@@ -70,6 +71,10 @@ libraries.junitJupiter = [
   "org.junit.jupiter:junit-jupiter-params:$versions.junitJupiter"
 ]
 
+libraries.wiremock = [
+  "com.github.tomakehurst:wiremock:$versions.wiremock"
+]
+
 libraries.jjwt = [
   "io.jsonwebtoken:jjwt:$versions.jjwt"
 ]
@@ -78,6 +83,17 @@ libraries.kafkaConnect = [
   "org.apache.kafka:connect-api:$versions.kafkaConnect"
 ]
 
+libraries.kafkaConnectRuntime = [
+  "org.apache.kafka:connect-runtime:$versions.kafkaConnect",
+  "org.apache.kafka:kafka_2.12:$versions.kafkaConnect",
+]
+
+libraries.kafkaConnectTest = [
+  "org.apache.kafka:connect-runtime:$versions.kafkaConnect:test",
+  "org.apache.kafka:kafka-clients:$versions.kafkaConnect:test",
+  "org.apache.kafka:kafka_2.12:$versions.kafkaConnect:test"
+]
+
 libraries.orgJson = [
   "org.json:json:$versions.orgJson"
 ]

streaming-connect-common/src/main/java/com/adobe/platform/streaming/AEPStreamingRuntimeException.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2021 Adobe. All rights reserved.
+ * This file is licensed to you under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License. You may obtain a copy
+ * of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under
+ * the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR REPRESENTATIONS
+ * OF ANY KIND, either express or implied. See the License for the specific language
+ * governing permissions and limitations under the License.
+ */
+
+package com.adobe.platform.streaming;
+
+import java.util.Optional;
+
+/**
+ * @author Adobe Inc.
+ */
+public class AEPStreamingRuntimeException extends RuntimeException {
+
+  private final Object errorContext;
+
+  public AEPStreamingRuntimeException(String message) {
+    super(message);
+    this.errorContext = null;
+  }
+
+  public AEPStreamingRuntimeException(String message, Throwable t) {
+    super(message, t);
+    this.errorContext = null;
+  }
+
+  public AEPStreamingRuntimeException(String message, Object errorContext) {
+    super(message);
+    this.errorContext = errorContext;
+  }
+
+  public AEPStreamingRuntimeException(String message, Throwable t, Object errorContext) {
+    super(message, t);
+    this.errorContext = errorContext;
+  }
+
+  public Optional<?> getErrorContext() {
+    return Optional.ofNullable(errorContext);
+  }
+}

streaming-connect-common/src/main/java/com/adobe/platform/streaming/http/HttpConnection.java

@@ -169,7 +169,7 @@ void close() {
           is.close();
         }
       } catch (IOException ioe) {
-        LOG.info("close(): failed to close input stream: {}", ioe.getMessage());
+        LOG.error("close(): failed to close input stream: {}", ioe.getMessage());
       }
 
       conn = null;

streaming-connect-common/src/main/java/com/adobe/platform/streaming/http/HttpUtil.java

@@ -12,6 +12,8 @@
 
 package com.adobe.platform.streaming.http;
 
+import com.google.common.annotations.VisibleForTesting;
+
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStream;
@@ -39,7 +41,7 @@ public static boolean isUnauthorized(int responseCode) {
     return responseCode == 401 || responseCode == 403;
   }
 
-  static String streamToString(InputStream in) throws HttpException {
+  public static String streamToString(InputStream in) throws HttpException {
     StringBuilder sb = new StringBuilder(128);
     try (BufferedReader r = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {
       String str;

streaming-connect-common/src/main/java/com/adobe/platform/streaming/http/serializer/SerializerDeserializerUtil.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2021 Adobe. All rights reserved.
+ * This file is licensed to you under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License. You may obtain a copy
+ * of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under
+ * the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR REPRESENTATIONS
+ * OF ANY KIND, either express or implied. See the License for the specific language
+ * governing permissions and limitations under the License.
+ */
+
+package com.adobe.platform.streaming.http.serializer;
+
+import com.adobe.platform.streaming.AEPStreamingRuntimeException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * @author Adobe Inc.
+ */
+public class SerializerDeserializerUtil {
+
+  private static final Logger errorLogger = LoggerFactory.getLogger(SerializerDeserializerUtil.class);
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  public static String serialize(Object value) {
+    try {
+      return MAPPER.writeValueAsString(value);
+    } catch (JsonProcessingException e) {
+      errorLogger.error("Failed to serialize value : {}", value);
+      throw new AEPStreamingRuntimeException("Failed to serialize value", value);
+    }
+  }
+
+  public static JsonNode convertToJsonNode(Object value) {
+    return MAPPER.convertValue(value, JsonNode.class);
+  }
+
+  public static JsonNode convertStringToJsonNode(Object value) {
+    try {
+      return MAPPER.readTree(value.toString());
+    } catch (IOException e) {
+      errorLogger.error("Failed to serialize value : {}", value);
+      throw new AEPStreamingRuntimeException("Failed to serialize value", value);
+    }
+  }
+
+  public static ObjectNode createObjectNode() {
+    return MAPPER.createObjectNode();
+  }
+}

streaming-connect-sink/build.gradle

@@ -23,8 +23,11 @@ dependencies {
                  libraries.jackson, libraries.httpClient,
                  libraries.kafkaConnect, libraries.orgJson,
                  libraries.slf4j
+  compileOnly libraries.kafkaConnectRuntime
   jmxAgent libraries.jmxPrometheusJavaAgent
-  testImplementation libraries.jmockit, libraries.junitJupiter
+  testImplementation libraries.jmockit, libraries.junitJupiter,
+                     libraries.wiremock, libraries.kafkaConnectTest,
+                     libraries.kafkaConnectRuntime
 }
 
 task copyJmxAgent(type: Copy) {

streaming-connect-sink/src/main/java/com/adobe/platform/streaming/sink/AbstractAEPPublisher.java

@@ -57,6 +57,8 @@ public abstract class AbstractAEPPublisher implements DataPublisher {
 
   private static final String AEP_CONNECTION_AUTH_ENABLED_VALUE = "true";
   private static final String AEP_CONNECTION_AUTH_DISABLED_VALUE = "false";
+  public static final String AEP_ERROR_LOGGER = "aep.error.logger";
+  public static final String AEP_ERROR_TOPIC = "aep.error.topic";
 
   protected HttpProducer getHttpProducer(Map<String, String> props) throws AEPStreamingException {
     return HttpProducer.newBuilder(getAepEndpoint(props.get(AEP_ENDPOINT)))

streaming-connect-sink/src/main/java/com/adobe/platform/streaming/sink/AbstractSinkTask.java

@@ -14,20 +14,27 @@
 
 import com.adobe.platform.streaming.AEPStreamingException;
 import com.adobe.platform.streaming.sink.utils.SinkUtils;
+
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
+
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.kafka.common.utils.AppInfoParser;
+import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.runtime.WorkerConfig;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
+import org.apache.kafka.connect.sink.SinkTaskContext;
+import org.codehaus.plexus.util.ReflectionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * @author Adobe Inc.
@@ -48,15 +55,25 @@ public abstract class AbstractSinkTask<T> extends SinkTask {
   private int flushIntervalMillis;
   private int flushBytesCount;
   private long lastFlushMilliSec = System.currentTimeMillis();
+  private String bootstrapServers;
 
   @Override
   public String version() {
     return AppInfoParser.getVersion();
   }
 
+  @Override
+  public void initialize(SinkTaskContext context) {
+    super.initialize(context);
+    bootstrapServers = getBootstrapServers(context);
+  }
+
   @Override
   public void start(Map<String, String> props) {
     LOG.info("Started Sink Task with props: {}", props);
+    if (Objects.nonNull(bootstrapServers)) {
+      props.put(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+    }
 
     try {
       flushIntervalMillis = SinkUtils.getProperty(props, FLUSH_INTERVAL_SECS, DEFAULT_FLUSH_INTERVAL, MILLIS_IN_A_SEC);
@@ -133,6 +150,19 @@ private void publishAndLogIfRequired(List<T> eventsToPublish) {
     }
   }
 
+  private String getBootstrapServers(SinkTaskContext context) {
+    try {
+      Object workerSinkTask = ReflectionUtils.getValueIncludingSuperclasses("sinkTask", context);
+      WorkerConfig workerConfig = (WorkerConfig) ReflectionUtils
+        .getValueIncludingSuperclasses("workerConfig", workerSinkTask);
+
+      return Utils.join(workerConfig.getList(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG), ",");
+    } catch (IllegalAccessException exception) {
+      LOG.error("Failed to get bootstrap server.", exception);
+    }
+    return null;
+  }
+
   private void reset(List<T> eventsToPublish, long tempCurrentTime) {
     lastFlushMilliSec = tempCurrentTime;
     bytesRead = 0;

streaming-connect-sink/src/main/java/com/adobe/platform/streaming/sink/impl/AEPPublisher.java

@@ -18,6 +18,8 @@
 import com.adobe.platform.streaming.http.HttpProducer;
 import com.adobe.platform.streaming.http.HttpUtil;
 import com.adobe.platform.streaming.sink.AbstractAEPPublisher;
+import com.adobe.platform.streaming.sink.impl.reporter.CompositeErrorReporter;
+import com.adobe.platform.streaming.sink.impl.reporter.ErrorReporter;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.http.entity.ContentType;
@@ -28,21 +30,24 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Future;
 
 /**
  * @author Adobe Inc.
  */
-class AEPPublisher extends AbstractAEPPublisher {
+public class AEPPublisher extends AbstractAEPPublisher {
 
   private static final Logger LOG = LoggerFactory.getLogger(AEPPublisher.class);
   private static final String MESSAGES_KEY = "messages";
 
   private int count;
   private final HttpProducer producer;
+  private final ErrorReporter<List<Future<?>>> errorReporter;
 
   AEPPublisher(Map<String, String> props) throws AEPStreamingException {
     count = 0;
     producer = getHttpProducer(props);
+    errorReporter = new CompositeErrorReporter(props);
   }
 
   @Override
@@ -69,6 +74,7 @@ public void publishData(List<String> messages) throws AEPStreamingException {
       LOG.debug("Successfully published data to Adobe Experience Platform: {}", response);
     } catch (HttpException httpException) {
       LOG.error("Failed to publish data to Adobe Experience Platform", httpException);
+      errorReporter.report(messages, httpException);
       if (HttpUtil.is500(httpException.getResponseCode()) || HttpUtil.isUnauthorized(httpException.getResponseCode())) {
         throw new AEPStreamingException("Failed to publish", httpException);
       }