[GitHub] rabbah commented on a change in pull request #3325: Refactor Kafka clients for more generalized configuration reading.

[GitBox <gi...@apache.org>]

rabbah commented on a change in pull request #3325: Refactor Kafka clients for more generalized configuration reading.
URL: https://github.com/apache/incubator-openwhisk/pull/3325#discussion_r175062886
 
 

 ##########
 File path: common/scala/src/main/scala/whisk/connector/kafka/KafkaProducerConnector.scala
 ##########
 @@ -93,35 +87,20 @@ class KafkaProducerConnector(kafkahosts: String, id: String = UUIDs.randomUUID()
 
   private val sentCounter = new Counter()
 
-  private def getProps: Properties = {
-    val props = new Properties
-    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkahosts)
+  private def createProducer(): KafkaProducer[String, String] = {
+    val config = Map(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> kafkahosts) ++
+      configMapToKafkaConfig(loadConfigOrThrow[Map[String, String]](ConfigKeys.kafkaCommon)) ++
+      configMapToKafkaConfig(loadConfigOrThrow[Map[String, String]](ConfigKeys.kafkaProducer))
 
-    // Load additional config from the config files and add them here.
-    val config =
-      KafkaConfiguration.configMapToKafkaConfig(loadConfigOrThrow[Map[String, String]](ConfigKeys.kafkaCommon)) ++
-        KafkaConfiguration.configMapToKafkaConfig(loadConfigOrThrow[Map[String, String]](ConfigKeys.kafkaProducer))
-
-    config.foreach {
-      case (key, value) => props.put(key, value)
-    }
-    props
-  }
-
-  private def getProducer(props: Properties): KafkaProducer[String, String] = {
-    val keySerializer = new StringSerializer
-    val valueSerializer = new StringSerializer
-    new KafkaProducer(props, keySerializer, valueSerializer)
+    new KafkaProducer(config, new StringSerializer, new StringSerializer)
   }
 
   private def recreateProducer(): Unit = {
     val oldProducer = producer
-    Future {
-      oldProducer.close()
-      logging.info(this, s"old consumer closed")
-    }
-    producer = getProducer(getProps)
+    oldProducer.close()
+    logging.info(this, s"old consumer closed")
 
 Review comment:
   Will gladly accept a pr :)

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services