Merge db7e4c8 into de3e0a8

Author: bdoyle0182

common/scala/src/main/scala/org/apache/openwhisk/connector/kafka/KafkaConsumerConnector.scala

@@ -55,6 +55,7 @@ class KafkaConsumerConnector(
 
   // Currently consumed offset, is used to calculate the topic lag.
   // It is updated from one thread in "peek", no concurrent data structure is necessary
+  // Note: Currently, this value used for metric reporting will not be accurate if using a multi-partition topic.
   private var offset: Long = 0
 
   // Markers for metrics, initialized only once

common/scala/src/main/scala/org/apache/openwhisk/connector/kafka/KafkaMessagingProvider.scala

@@ -64,7 +64,7 @@ object KafkaMessagingProvider extends MessagingProvider {
 
     Try(AdminClient.create(commonConfig + (AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG -> config.kafkaHosts)))
       .flatMap(client => {
-        val partitions = 1
+        val partitions = topicConfig.getOrElse("partitions", "1").toInt
         val nt = new NewTopic(topic, partitions, kafkaConfig.replicationFactor).configs(topicConfig.asJava)
 
         def createTopic(retries: Int = 5): Try[Unit] = {

common/scala/src/main/scala/org/apache/openwhisk/connector/kafka/KafkaProducerConnector.scala

@@ -51,7 +51,7 @@ class KafkaProducerConnector(
   /** Sends msg to topic. This is an asynchronous operation. */
   override def send(topic: String, msg: Message, retry: Int = 3): Future[ResultMetadata] = {
     implicit val transid: TransactionId = msg.transid
-    val record = new ProducerRecord[String, String](topic, "messages", msg.serialize)
+    val record = new ProducerRecord[String, String](topic, msg.serialize)
     val produced = Promise[ResultMetadata]()
 
     Future {