bump kafka dep. to 10.0.2.1. also bump testng (linkedin#64)

becketqin · Sep 20, 2017 · 0d2f1de · 0d2f1de
1 parent 6e6309a
commit 0d2f1de
Show file tree

Hide file tree

Showing 4 changed files with 23 additions and 9 deletions.
diff --git a/build.gradle b/build.gradle
@@ -46,12 +46,12 @@ repositories {
 }
 
 dependencies {
-    compile 'org.apache.kafka:kafka-clients:0.10.1.1'
+    compile 'org.apache.kafka:kafka-clients:0.10.2.1'
 
-    testCompile 'org.apache.kafka:kafka_2.10:0.10.1.1:test'
-    testCompile 'org.apache.kafka:kafka_2.10:0.10.1.1'
+    testCompile 'org.apache.kafka:kafka_2.10:0.10.2.1:test'
+    testCompile 'org.apache.kafka:kafka_2.10:0.10.2.1'
     testCompile 'org.bouncycastle:bcpkix-jdk15on:1.54'
-    testCompile 'org.testng:testng:6.8.8'
+    testCompile 'org.testng:testng:6.11'
 }
 
 //test configurations

diff --git a/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaConsumerImpl.java b/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaConsumerImpl.java
@@ -15,6 +15,7 @@
 import com.linkedin.kafka.clients.utils.LiKafkaClientsUtils;
 import java.util.Collections;
 import java.util.Locale;
+import java.util.concurrent.TimeUnit;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
@@ -535,6 +536,15 @@ public void close() {
     _consumerRecordsProcessor.close();
   }
 
+  @Override
+  public void close(long timeout, TimeUnit unit) {
+    if (_autoCommitEnabled) {
+      commitSync();
+    }
+    _kafkaConsumer.close(timeout, unit);
+    _consumerRecordsProcessor.close();
+  }
+
   @Override
   public void wakeup() {
     _kafkaConsumer.wakeup();

diff --git a/src/test/java/com/linkedin/kafka/clients/auditing/abstractimpl/AbstractAuditorTest.java b/src/test/java/com/linkedin/kafka/clients/auditing/abstractimpl/AbstractAuditorTest.java
@@ -275,6 +275,11 @@ public long milliseconds() {
       return now;
     }
 
+    @Override
+    public long hiResClockMs() {
+      return milliseconds();
+    }
+
     @Override
     public long nanoseconds() {
       // Not used.

diff --git a/src/test/scala/com/linkedin/kafka/clients/utils/tests/TestUtils.scala b/src/test/scala/com/linkedin/kafka/clients/utils/tests/TestUtils.scala
@@ -9,10 +9,9 @@ import java.nio.file.Files
 import java.util.{Properties, Random}
 
 import kafka.server.{KafkaConfig, KafkaServer}
-import kafka.utils.{SystemTime, Time}
-import org.apache.kafka.common.network.Mode
+import org.apache.kafka.common.network.{ListenerName, Mode}
 import org.apache.kafka.common.protocol.SecurityProtocol
-import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.utils.{Time, Utils}
 import org.apache.kafka.common.utils.Utils.formatAddress
 
 import scala.collection.JavaConversions._
@@ -34,14 +33,14 @@ object TestUtils {
    *
    * @param config The configuration of the server
    */
-  def createServer(config: KafkaConfig, time: Time = SystemTime): KafkaServer = {
+  def createServer(config: KafkaConfig, time: Time = Time.SYSTEM): KafkaServer = {
     val server = new KafkaServer(config, time)
     server.startup()
     server
   }
 
   def getBrokerListStrFromServers(servers: Seq[KafkaServer], protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT): String = {
-    servers.map(s => formatAddress(s.config.hostName, s.boundPort(protocol))).mkString(",")
+    servers.map(s => formatAddress(s.config.hostName, s.boundPort(ListenerName.forSecurityProtocol(protocol)))).mkString(",")
   }
 
   /**