Add custom endpoint overrides for Kinesis and DynamoDB (closes #157)

Author: ZAYEC77

core/src/main/scala/com.snowplowanalytics.stream/loader/Config.scala

@@ -84,8 +84,10 @@ object Config {
       initialTimestamp: Option[String],
       maxRecords: Long,
       region: String,
-      appName: String)
-        extends Queue {
+      appName: String,
+      customEndpoint: Option[String],
+      dynamodbCustomEndpoint: Option[String]
+    ) extends Queue {
       val timestampEither = initialTimestamp
         .toRight("An initial timestamp needs to be provided when choosing AT_TIMESTAMP")
         .right
@@ -99,10 +101,15 @@ object Config {
 
       val timestamp = timestampEither.right.toOption
 
-      val endpoint = region match {
+      val endpoint = customEndpoint.getOrElse(region match {
         case cn @ "cn-north-1" => s"https://kinesis.$cn.amazonaws.com.cn"
         case _                 => s"https://kinesis.$region.amazonaws.com"
-      }
+      })
+
+      val dynamodbEndpoint = dynamodbCustomEndpoint.getOrElse(region match {
+        case cn @ "cn-north-1" => s"https://dynamodb.$cn.amazonaws.com.cn"
+        case _                 => s"https://dynamodb.$region.amazonaws.com"
+      })
     }
   }
 

core/src/main/scala/com.snowplowanalytics.stream/loader/EsLoaderBadRow.scala

@@ -40,11 +40,13 @@ case class EsLoaderBadRow(line: String, errors: NonEmptyList[String]) {
     .withZone(DateTimeZone.UTC)
 
   def toCompactJson =
-    Json.obj(
-      "line"           -> line.asJson,
-      "errors"         -> errors.asJson,
-      "failure_tstamp" -> getTstamp(tstamp, tstampFormat).asJson
-    ).noSpaces
+    Json
+      .obj(
+        "line"           -> line.asJson,
+        "errors"         -> errors.asJson,
+        "failure_tstamp" -> getTstamp(tstamp, tstampFormat).asJson
+      )
+      .noSpaces
 }
 
 object EsLoaderBadRow {

core/src/main/scala/com.snowplowanalytics.stream/loader/executors/KinesisSourceExecutor.scala

@@ -67,6 +67,7 @@ class KinesisSourceExecutor[A, B](
       kcc.AWS_CREDENTIALS_PROVIDER,
       kcc.WORKER_ID)
       .withKinesisEndpoint(kcc.KINESIS_ENDPOINT)
+      .withDynamoDBEndpoint(kcc.DYNAMODB_ENDPOINT)
       .withFailoverTimeMillis(kcc.FAILOVER_TIME)
       .withMaxRecords(kcc.MAX_RECORDS)
       .withIdleTimeBetweenReadsInMillis(kcc.IDLE_TIME_BETWEEN_READS)
@@ -101,6 +102,7 @@ class KinesisSourceExecutor[A, B](
     queue: Queue.Kinesis): KinesisConnectorConfiguration = {
     val props = new Properties
     props.setProperty(KinesisConnectorConfiguration.PROP_KINESIS_ENDPOINT, queue.endpoint)
+    props.setProperty(KinesisConnectorConfiguration.PROP_DYNAMODB_ENDPOINT, queue.dynamodbEndpoint)
     props.setProperty(KinesisConnectorConfiguration.PROP_APP_NAME, queue.appName.trim)
     props.setProperty(
       KinesisConnectorConfiguration.PROP_INITIAL_POSITION_IN_STREAM,

examples/config.hocon.sample

@@ -78,6 +78,16 @@ queue {
   # Region where the Kinesis stream is located
   region = "{{kinesisRegion}}"
 
+  # Optional endpoint url configuration to override aws kinesis endpoints,
+  # this can be used to specify local endpoints when using localstack
+  # customEndpoint = {{kinesisEndpoint}}
+  # customEndpoint = ${?ENRICH_STREAMS_SOURCE_SINK_CUSTOM_ENDPOINT}
+
+  # Optional endpoint url configuration to override aws dyanomdb endpoints for Kinesis checkpoints lease table,
+  # this can be used to specify local endpoints when using Localstack
+  # dynamodbCustomEndpoint = "http://localhost:4569"
+  # dynamodbCustomEndpoint = ${?ENRICH_DYNAMODB_CUSTOM_ENDPOINT}
+
   # "appName" is used for a DynamoDB table to maintain stream state.
   # You can set it automatically using: "SnowplowElasticsearchSink-${sink.kinesis.in.stream-name}"
   appName = "{{kinesisAppName}}"