address configuration changes needed in lamba-datastore

Author: Neil Buesing

geomesa-lambda/geomesa-lambda-datastore/src/main/scala/org/locationtech/geomesa/lambda/data/LambdaDataStore.scala

@@ -218,5 +218,6 @@ object LambdaDataStore {
       consumers: Int,
       expiry: Option[FiniteDuration],
       persistBatchSize: Option[Int] = None,
+      offsetCommitIntervalMs: Long = 10000
     )
 }

geomesa-lambda/geomesa-lambda-datastore/src/main/scala/org/locationtech/geomesa/lambda/data/LambdaDataStoreParams.scala

@@ -104,6 +104,13 @@ object LambdaDataStoreParams extends GeoMesaDataStoreParams with SecurityParams
       "Offset manager instance to use",
       deprecatedKeys = Seq("lamdab.offset-manager", "offsetManager"))
 
+  val ConsumerOffsetCommitIntervalMs =
+    new GeoMesaParam[java.lang.Long](
+      "lambda.kafka.consumer.offset-commit-interval-ms",
+      "The frequency of committing offsets for the Kafka consumer",
+      default = 10000
+    )
+
   def parse(params: java.util.Map[String, _], namespace: String): LambdaConfig = {
     val brokers = BrokersParam.lookup(params)
     val expiry = if (!PersistParam.lookup(params).booleanValue) { None } else {
@@ -121,6 +128,8 @@ object LambdaDataStoreParams extends GeoMesaDataStoreParams with SecurityParams
     val zk = ZookeepersParam.lookup(params)
     val zkNamespace = s"gm_lambda_$namespace"
 
-    LambdaConfig(zk, zkNamespace, producerConfig, consumerConfig, partitions, consumers, expiry, batchSize)
+    val offsetCommitIntervalMs : Long = ConsumerOffsetCommitIntervalMs.lookupOpt(params).map(_.toLong).getOrElse(10000)
+
+    LambdaConfig(zk, zkNamespace, producerConfig, consumerConfig, partitions, consumers, expiry, batchSize, offsetCommitIntervalMs)
   }
 }

geomesa-lambda/geomesa-lambda-datastore/src/main/scala/org/locationtech/geomesa/lambda/stream/kafka/KafkaCacheLoader.scala

@@ -33,7 +33,8 @@ class KafkaCacheLoader(
     topic: String,
     frequency: Long,
     serializer: KryoFeatureSerializer,
-    cache: WritableFeatureCache) extends ThreadedConsumer(consumers, Duration.ofMillis(frequency)) {
+    cache: WritableFeatureCache,
+    offsetCommitIntervalMs: Long) extends ThreadedConsumer(consumers, Duration.ofMillis(frequency), offsetCommitIntervalMs) {
 
   startConsumers()
 

geomesa-lambda/geomesa-lambda-datastore/src/main/scala/org/locationtech/geomesa/lambda/stream/kafka/KafkaStore.scala

@@ -70,7 +70,7 @@ class KafkaStore(
   private val loader = {
     val consumers = KafkaStore.consumers(config.consumerConfig, topic, offsetManager, config.consumers, cache.partitionAssigned)
     val frequency = KafkaStore.LoadIntervalProperty.toDuration.get.toMillis
-    new KafkaCacheLoader(consumers, topic, frequency, serializer, cache)
+    new KafkaCacheLoader(consumers, topic, frequency, serializer, cache, config.offsetCommitIntervalMs)
   }
 
   override def createSchema(): Unit = {