Upgrade to kafka 2.0.0 (linkedin#132)

Author: lindong28

build.gradle

@@ -17,8 +17,8 @@ allprojects {
     compile 'net.sourceforge.argparse4j:argparse4j:0.5.0'
     compile 'org.slf4j:slf4j-log4j12:1.7.6'
     compile 'org.apache.avro:avro:1.4.0'
-    compile 'org.apache.kafka:kafka_2.12:1.1.1'
-    compile 'org.apache.kafka:kafka-clients:1.1.1'
+    compile 'org.apache.kafka:kafka_2.12:2.0.0'
+    compile 'org.apache.kafka:kafka-clients:2.0.0'
     compile 'org.testng:testng:6.8.8'
     compile 'org.eclipse.jetty:jetty-server:8.1.19.v20160209'
     compile 'org.json:json:20140107'

src/main/java/com/linkedin/kmf/consumer/OldConsumer.java

@@ -14,7 +14,6 @@
 import com.linkedin.kmf.consumer.BaseConsumerRecord;
 import com.linkedin.kmf.consumer.KMBaseConsumer;
 import com.linkedin.kmf.consumer.NewConsumer;
-import com.linkedin.kmf.consumer.OldConsumer;
 import com.linkedin.kmf.services.configs.ConsumeServiceConfig;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -93,11 +92,6 @@ public ConsumeService(Map<String, Object> props, String name) throws Exception {
 
     if (consumerClassName.equals(NewConsumer.class.getCanonicalName()) || consumerClassName.equals(NewConsumer.class.getSimpleName())) {
       consumerClassName = NewConsumer.class.getCanonicalName();
-    } else if (consumerClassName.equals(OldConsumer.class.getCanonicalName()) || consumerClassName.equals(OldConsumer.class.getSimpleName())) {
-      consumerClassName = OldConsumer.class.getCanonicalName();
-      // The name/value of these configs are changed in the new consumer.
-      consumerProps.put("auto.commit.enable", "false");
-      consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "largest");
     }
 
     // Assign config specified for ConsumeService.

src/main/java/com/linkedin/kmf/services/ConsumeService.java

@@ -29,7 +29,7 @@ public class ConsumeServiceConfig extends AbstractConfig {
 
   public static final String CONSUMER_CLASS_CONFIG = "consume.consumer.class";
   public static final String CONSUMER_CLASS_DOC = "Consumer class that will be instantiated as consumer in the consume service. "
-    + "It can be NewConsumer, OldConsumer, or full class name of any class that implements the KMBaseConsumer interface.";
+    + "It can be NewConsumer or full class name of any class that implements the KMBaseConsumer interface.";
 
   public static final String LATENCY_PERCENTILE_MAX_MS_CONFIG = "consume.latency.percentile.max.ms";
   public static final String LATENCY_PERCENTILE_MAX_MS_DOC = "This is used to derive the bucket number used to configure latency percentile metric. "