GEOMESA-3375 Kafka - add consumer props to admin client connection (#…

…3134) * Consumer props may contain connection parameters (SASL) set within GeoServer
locationtech · Jul 9, 2024 · 0111826 · 0111826
1 parent 638b631
commit 0111826
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 0 deletions.
diff --git a/...a-kafka-datastore/src/main/scala/org/locationtech/geomesa/kafka/data/KafkaDataStore.scala b/...a-kafka-datastore/src/main/scala/org/locationtech/geomesa/kafka/data/KafkaDataStore.scala
@@ -229,6 +229,7 @@ class KafkaDataStore(
     val topic = KafkaDataStore.topic(sft)
     val props = new Properties()
     props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.brokers)
+    config.consumers.properties.foreach { case (k, v) => props.put(k, v) }
     config.producers.properties.foreach { case (k, v) => props.put(k, v) }
 
     WithClose(AdminClient.create(props)) { admin =>
@@ -266,6 +267,7 @@ class KafkaDataStore(
     val topic = KafkaDataStore.topic(sft)
     val props = new Properties()
     props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.brokers)
+    config.consumers.properties.foreach { case (k, v) => props.put(k, v) }
     config.producers.properties.foreach { case (k, v) => props.put(k, v) }
 
     WithClose(AdminClient.create(props)) { admin =>

diff --git a/...sa-kafka-datastore/src/main/scala/org/locationtech/geomesa/kafka/data/KafkaMetadata.scala b/...sa-kafka-datastore/src/main/scala/org/locationtech/geomesa/kafka/data/KafkaMetadata.scala
@@ -111,6 +111,7 @@ class KafkaMetadata[T](val config: KafkaDataStoreConfig, val serializer: Metadat
   private def adminClientOp[V](fn: AdminClient => V): V = {
     val props = new Properties()
     props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.brokers)
+    config.consumers.properties.foreach { case (k, v) => props.put(k, v) }
     config.producers.properties.foreach { case (k, v) => props.put(k, v) }
     WithClose(AdminClient.create(props)) { admin => fn(admin) }
   }