Update hibench common

* Added profile scala 2.12 to POM. Scala < 2.11 does not compiles on
java 1.11 jdk.
* Changed in POM org.apache.kafka to 0.10.2.2.
* Following scala files has been rewritten to work with kafka_2.12
version 0.10.2.2:
	* KafkaCollector.scala
	* KafkaConsumer.scala
	* MetricsUtil.scala

Signed-off-by: Luis Ponce <luis.f.ponce.navarro@linux.intel.com>

Author: luisfponce

common/pom.xml

@@ -41,7 +41,7 @@
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka_${scala.binary.version}</artifactId>
-      <version>0.8.2.1</version>
+      <version>0.10.2.2</version>
     </dependency>
   </dependencies>
 
@@ -134,5 +134,20 @@
         </property>
       </activation>
     </profile>
+
+    <profile>
+      <id>scala2.12</id>
+      <properties>
+        <scala.version>2.12.8</scala.version>
+        <scala.binary.version>2.12</scala.binary.version>
+      </properties>
+      <activation>
+        <property>
+          <name>scala</name>
+          <value>2.12</value>
+        </property>
+      </activation>
+    </profile>
+
   </profiles>
 </project>

common/src/main/scala/com/intel/hibench/common/streaming/metrics/KafkaCollector.scala

@@ -22,7 +22,7 @@ import java.util.Date
 import java.util.concurrent.{TimeUnit, Future, Executors}
 
 import com.codahale.metrics.{UniformReservoir, Histogram}
-import kafka.utils.{ZKStringSerializer, ZkUtils}
+import kafka.utils.{ZKStringSerializer$, ZkUtils}
 import org.I0Itec.zkclient.ZkClient
 
 import scala.collection.mutable.ArrayBuffer
@@ -60,9 +60,10 @@ class KafkaCollector(zkConnect: String, metricsTopic: String,
   }
 
   private def getPartitions(topic: String, zkConnect: String): Seq[Int] = {
-    val zkClient = new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer)
+    val zkClient = new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer$.MODULE$)
+    val zkUtils = ZkUtils.apply(zkClient, false);
     try {
-      ZkUtils.getPartitionsForTopics(zkClient, Seq(topic)).flatMap(_._2).toSeq
+      zkUtils.getPartitionsForTopics(Seq(topic)).flatMap(_._2).toSeq
     } finally {
       zkClient.close()
     }

common/src/main/scala/com/intel/hibench/common/streaming/metrics/KafkaConsumer.scala

@@ -23,8 +23,11 @@ import kafka.common.ErrorMapping._
 import kafka.common.TopicAndPartition
 import kafka.consumer.{ConsumerConfig, SimpleConsumer}
 import kafka.message.MessageAndOffset
-import kafka.utils.{ZKStringSerializer, ZkUtils, Utils}
+import kafka.utils.{ZKStringSerializer$, ZkUtils}
 import org.I0Itec.zkclient.ZkClient
+import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.protocol.SecurityProtocol
+import org.apache.kafka.common.network.ListenerName
 
 class KafkaConsumer(zookeeperConnect: String, topic: String, partition: Int) {
 
@@ -67,15 +70,15 @@ class KafkaConsumer(zookeeperConnect: String, topic: String, partition: Int) {
   }
 
   private def createConsumer: SimpleConsumer = {
-    val zkClient = new ZkClient(zookeeperConnect, 6000, 6000, ZKStringSerializer)
+    val zkClient = new ZkClient(zookeeperConnect, 6000, 6000, ZKStringSerializer$.MODULE$)
+    val zkUtils = ZkUtils.apply(zkClient, false);
     try {
-      val leader = ZkUtils.getLeaderForPartition(zkClient, topic, partition)
-          .getOrElse(throw new RuntimeException(
-            s"leader not available for TopicAndPartition($topic, $partition)"))
-      val broker = ZkUtils.getBrokerInfo(zkClient, leader)
-          .getOrElse(throw new RuntimeException(s"broker info not found for leader $leader"))
-      new SimpleConsumer(broker.host, broker.port,
-        config.socketTimeoutMs, config.socketReceiveBufferBytes, CLIENT_ID)
+      val leader = zkUtils.getLeaderForPartition(topic, partition)
+        .getOrElse(throw new RuntimeException(s"leader not available for TopicAndPartition($topic, $partition)"))
+      val brokerInfo = zkUtils.getBrokerInfo(leader)
+        .getOrElse(throw new RuntimeException(s"broker info not found for leader $leader"))
+      val broker = brokerInfo.getBrokerEndPoint(ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))
+      new SimpleConsumer(broker.host, broker.port , config.socketTimeoutMs, config.socketReceiveBufferBytes, CLIENT_ID)
     } catch {
       case e: Exception =>
         throw e

common/src/main/scala/com/intel/hibench/common/streaming/metrics/MetricsUtil.scala

@@ -18,7 +18,8 @@ package com.intel.hibench.common.streaming.metrics
 
 import com.intel.hibench.common.streaming.Platform
 import kafka.admin.AdminUtils
-import kafka.utils.ZKStringSerializer
+import kafka.utils.ZKStringSerializer$
+import kafka.utils.ZkUtils
 import org.I0Itec.zkclient.ZkClient
 
 object MetricsUtil {
@@ -34,10 +35,11 @@ object MetricsUtil {
   }
 
   def createTopic(zkConnect: String, topic: String, partitions: Int): Unit = {
-    val zkClient = new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer)
+    val zkClient = new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer$.MODULE$)
     try {
-      AdminUtils.createTopic(zkClient, topic, partitions, 1)
-      while (!AdminUtils.topicExists(zkClient, topic)) {
+      val zkUtils = ZkUtils.apply(zkClient, false);
+      AdminUtils.createTopic(zkUtils, topic, partitions, 1)
+      while (!AdminUtils.topicExists(zkUtils, topic)) {
         Thread.sleep(100)
       }
     } catch {