adding dynamo read write throughput

deployment/terraform/kinesis-span-collector/main.tf

@@ -27,6 +27,8 @@ data "template_file" "config_data" {
     kafka_endpoint = "${var.kafka_endpoint}"
     sts_role_arn = "${var.sts_role_arn}"
     app_group_name = "${var.haystack_cluster_name}-${var.app_name}"
+    dynamodb_read_thoroughput = "${var.dynamodb_read_throughput}"
+    dynamodb_write_throughput = "${var.dynamodb_write_throughput}"
   }
 }
 

deployment/terraform/kinesis-span-collector/templates/kinesis-span-collector_conf.tpl

@@ -27,6 +27,11 @@ kinesis {
     retry.interval.ms = 250
   }
 
+  dynamodb {
+      readThroughput = ${dynamodb_read_throughput}
+      writeThroughput = ${dynamodb_write_throughput}
+    }
+
   task.backoff.ms = 200
   max.records.read = 2000
   idle.time.between.reads.ms = 500

deployment/terraform/kinesis-span-collector/variables.tf

@@ -18,6 +18,8 @@ variable "cpu_request"{}
 variable "cpu_limit"{}
 variable "app_name"{ default = "kinesis-span-collector" }
 variable "env_vars" {}
+variable "dynamodb_read_throughput" {}
+variable "dynamodb_write_throughput" {}
 
 variable "termination_grace_period" {
   default = 30

deployment/terraform/main.tf

@@ -22,6 +22,8 @@ module "kinesis-span-collector" {
   memory_limit = "${var.collector["kinesis_span_collector_memory_limit"]}"
   jvm_memory_limit = "${var.collector["kinesis_span_collector_jvm_memory_limit"]}"
   app_name = "${var.collector["kinesis_span_collector_app_name"]}"
+  dynamodb_read_throughput = "${var.collector["kinesis_span_collector_dynamodb_read_thoroughput"]}"
+  dynamodb_write_throughput = "${var.collector["kinesis_span_collector_dynamodb_write_thoroughput"]}"
 }
 
 module "http-span-collector" {

kinesis/src/main/resources/config/base.conf

@@ -33,6 +33,11 @@ kinesis {
     retry.interval.ms = 250
   }
 
+  dynamodb {
+    readThroughput = 10
+    writeThroughput = 10
+  }
+
   task.backoff.ms = 200
   max.records.read = 2000
   idle.time.between.reads.ms = 500

kinesis/src/main/scala/com/expedia/www/haystack/kinesis/span/collector/config/ProjectConfiguration.scala

@@ -52,6 +52,8 @@ object ProjectConfiguration {
       kinesisEndpoint = if (kinesis.hasPath("endpoint")) Some(kinesis.getString("endpoint")) else None,
       dynamoEndpoint = if (kinesis.hasPath("dynamodb.endpoint")) Some(kinesis.getString("dynamodb.endpoint")) else None,
       dynamoTableName = if(kinesis.hasPath("dynamodb.table")) Some(kinesis.getString("dynamodb.table")) else None,
+      dynamoReadThroughput = if (kinesis.hasPath("dynamodb.readThroughput")) Some(kinesis.getInt("dynamodb.readThroughput")) else None,
+      dynamoWriteThroughput = if (kinesis.hasPath("dynamodb.writeThroughput")) Some(kinesis.getInt("dynamodb.writeThroughput")) else None,
       maxRecordsToRead = kinesis.getInt("max.records.read"),
       idleTimeBetweenReads = kinesis.getDuration("idle.time.between.reads.ms", TimeUnit.MILLISECONDS).millis,
       shardSyncInterval = kinesis.getDuration("shard.sync.interval.ms", TimeUnit.MILLISECONDS).millis,

kinesis/src/main/scala/com/expedia/www/haystack/kinesis/span/collector/config/entities/KinesisConsumerConfiguration.scala

@@ -33,6 +33,8 @@ case class KinesisConsumerConfiguration(awsRegion: String,
                                         kinesisEndpoint: Option[String],
                                         dynamoEndpoint: Option[String],
                                         dynamoTableName: Option[String],
+                                        dynamoReadThroughput: Option[Int],
+                                        dynamoWriteThroughput: Option[Int],
                                         maxRecordsToRead: Int,
                                         idleTimeBetweenReads: FiniteDuration,
                                         shardSyncInterval: FiniteDuration,

kinesis/src/main/scala/com/expedia/www/haystack/kinesis/span/collector/kinesis/client/KinesisConsumer.scala

@@ -96,6 +96,8 @@ class KinesisConsumer(config: KinesisConsumerConfiguration,
 
     config.dynamoEndpoint.map(kinesisClientConfig.withDynamoDBEndpoint)
     config.kinesisEndpoint.map(kinesisClientConfig.withKinesisEndpoint)
+    config.dynamoReadThroughput.map(kinesisClientConfig.withInitialLeaseTableReadCapacity)
+    config.dynamoWriteThroughput.map(kinesisClientConfig.withInitialLeaseTableWriteCapacity)
 
     new Worker.Builder()
       .config(kinesisClientConfig)

kinesis/src/test/scala/com/expedia/www/haystack/kinesis/span/collector/unit/tests/RecordProcessorSpec.scala

@@ -47,7 +47,7 @@ class RecordProcessorSpec extends FunSpec with Matchers with EasyMockSugar with
   describe("Record Processor") {
 
     val kinesisConfig = KinesisConsumerConfiguration("us-west-2", None,
-      "app-group", "stream-1", InitialPositionInStream.LATEST, 10.seconds, 10, 10.seconds, None, None, None,
+      "app-group", "stream-1", InitialPositionInStream.LATEST, 10.seconds, 10, 10.seconds, None, None, None, None, None,
       10000, 500.millis, 10000.millis, MetricsLevel.NONE, 10000.millis, 200.millis)
 
     it("should process the record, sends to sink and perform checkpointing") {