Add ability to disable CloudWatch metrics (closes #160)

Author: jbeemster

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

@@ -86,7 +86,8 @@ object Config {
       region: String,
       appName: String,
       customEndpoint: Option[String],
-      dynamodbCustomEndpoint: Option[String]
+      dynamodbCustomEndpoint: Option[String],
+      disableCloudWatch: Option[Boolean]
     ) extends Queue {
       val timestampEither = initialTimestamp
         .toRight("An initial timestamp needs to be provided when choosing AT_TIMESTAMP")

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

@@ -38,6 +38,7 @@ import com.amazonaws.services.kinesis.connectors.interfaces.IKinesisConnectorPip
 import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration
 import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker
 import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory
+import com.amazonaws.services.kinesis.metrics.impl.NullMetricsFactory
 
 // This project
 import com.snowplowanalytics.stream.loader.Config._
@@ -166,17 +167,20 @@ class KinesisSourceExecutor[A, B](
         "idleTimeBetweenReads is greater than bufferTimeMillisecondsLimit. For best results, ensure that bufferTimeMillisecondsLimit is more than or equal to idleTimeBetweenReads ")
     }
 
-    val workerBuilder = new Worker.Builder()
-      .recordProcessorFactory(getKinesisConnectorRecordProcessorFactory())
-      .config(kinesisClientLibConfiguration)
-
-    // If a metrics factory was specified, use it.
-    if (metricFactory != null) {
-      workerBuilder.metricsFactory(metricFactory)
+    worker = kinesis.disableCloudWatch match {
+      case Some(true) =>
+        new Worker.Builder()
+          .recordProcessorFactory(getKinesisConnectorRecordProcessorFactory())
+          .config(kinesisClientLibConfiguration)
+          .metricsFactory(new NullMetricsFactory())
+          .build()
+      case _ =>
+        new Worker.Builder()
+          .recordProcessorFactory(getKinesisConnectorRecordProcessorFactory())
+          .config(kinesisClientLibConfiguration)
+          .build()
     }
 
-    worker = workerBuilder.build()
-
     LOG.info(getClass.getSimpleName + " worker created")
   }
 

examples/config.hocon.sample

@@ -88,6 +88,10 @@ queue {
   # dynamodbCustomEndpoint = "http://localhost:4569"
   # dynamodbCustomEndpoint = ${?ENRICH_DYNAMODB_CUSTOM_ENDPOINT}
 
+  # Optional override to disable cloudwatch
+  # disableCloudWatch = true
+  # disableCloudWatch = ${?ENRICH_DISABLE_CLOUDWATCH}
+
   # "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}}"