apache kafka scala version 2.13 with code version 2.4.1 (#251)

apache.kafka use version : '2.4.1'
justification: the internal kmf multiproduct currently employs scala version _2.11.
Moreover, Anasi (or ELR tool) doesn't allow for intaking scala _2.11 with apache kafka code version 2.5.0 because the maven service doesn't have, and thus does not release, the scala _2.11 version 2.5.0 .
we cannot bump up to scala version _2.13 in the kafka-monitoring multiproduct because there are many, many multiproduct dependencies on the scala version _2.11 that have not been ELR.ed yet.
Justifications for merging this:

the PR won’t break any dependencies because the bump to 2.5.0 happened very recently in this repository
2 - internal MP (xinfra monitor) can be bumped up to scala kafka_2.12 but even that requires modifications to several kmf-depended multiproducts, including trackingrest, container, etc.. which could easily take a long time to all complete.
bumping up kmf internal mp to scala 2.12 requires these changes which could take some time to code-complete, merge, and get released.

What went wrong:
Execution failed for task ':likafka-monitoring:kafka-monitoring-impl_2.12:compileJava'.
Could not resolve all files for configuration ':likafka-monitoring:kafka-monitoring-impl_2.12:compileClasspath'.
Could not find com.linkedin.linkedin-kafka-clients:linkedin-kafka-clients-message-processor_2.12:8.0.204.
Required by:
project :likafka-monitoring:kafka-monitoring-impl_2.12 > com.linkedin.kafka-rest:tracking-rest-impl:0.3.150
Could not find com.linkedin.container:tracker-consumer-internal_2.12:36.12.6.
Required by:
project :likafka-monitoring:kafka-monitoring-impl_2.12 > com.linkedin.kafka-rest:tracking-rest-impl:0.3.150 > com.linkedin.container:tracker-consumer-api:37.1.63
Could not find com.linkedin.likafka-cruise-control:likafka-cruise-control-impl_2.12:1.0.17.
Required by:
project :likafka-monitoring:kafka-monitoring-impl_2.12 > com.linkedin.kafka-rest:tracking-rest-impl:0.3.150 > com.linkedin.kafka-server:kafka-impl:2.0.0.71

lets have kmf mp stay in 2.11 for now.

build.gradle

@@ -38,8 +38,8 @@ allprojects {
     compile 'net.savantly:graphite-client:1.1.0-RELEASE'
     compile 'com.timgroup:java-statsd-client:3.0.1'
     compile 'com.signalfx.public:signalfx-codahale:0.0.47'
-    compile group: 'org.apache.kafka', name: 'kafka_2.13', version: '2.5.0'
-    compile group: 'org.apache.kafka', name: 'kafka-clients', version: '2.5.0'
+    compile group: 'org.apache.kafka', name: 'kafka_2.13', version: '2.4.1'
+    compile group: 'org.apache.kafka', name: 'kafka-clients', version: '2.4.1'
     testCompile 'org.mockito:mockito-core:2.24.0'
     testCompile 'org.testng:testng:6.8.8'
   }

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

@@ -54,7 +54,6 @@
 import org.apache.kafka.common.utils.Time;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import scala.Option;
 import scala.Option$;
 import scala.collection.Seq;
 
@@ -388,9 +387,9 @@ private Set<Node> getAvailableBrokers() throws ExecutionException, InterruptedEx
     }
 
     void maybeReassignPartitionAndElectLeader() throws Exception {
-      try (KafkaZkClient zkClient = KafkaZkClient.apply(_zkConnect, JaasUtils.isZkSaslEnabled(),
+      try (KafkaZkClient zkClient = KafkaZkClient.apply(_zkConnect, JaasUtils.isZkSecurityEnabled(),
           com.linkedin.kmf.common.Utils.ZK_SESSION_TIMEOUT_MS, com.linkedin.kmf.common.Utils.ZK_CONNECTION_TIMEOUT_MS,
-          Integer.MAX_VALUE, Time.SYSTEM, METRIC_GROUP_NAME, "SessionExpireListener", null, Option.apply(null))) {
+          Integer.MAX_VALUE, Time.SYSTEM, METRIC_GROUP_NAME, "SessionExpireListener", null)) {
 
         List<TopicPartitionInfo> partitionInfoList = _adminClient
             .describeTopics(Collections.singleton(_topic)).all().get().get(_topic).partitions();
@@ -463,8 +462,8 @@ void maybeElectLeader() throws Exception {
         return;
       }
 
-      try (KafkaZkClient zkClient = KafkaZkClient.apply(_zkConnect, JaasUtils.isZkSaslEnabled(), com.linkedin.kmf.common.Utils.ZK_SESSION_TIMEOUT_MS,
-          com.linkedin.kmf.common.Utils.ZK_CONNECTION_TIMEOUT_MS, Integer.MAX_VALUE, Time.SYSTEM, METRIC_GROUP_NAME, "SessionExpireListener", null, null)) {
+      try (KafkaZkClient zkClient = KafkaZkClient.apply(_zkConnect, JaasUtils.isZkSecurityEnabled(), com.linkedin.kmf.common.Utils.ZK_SESSION_TIMEOUT_MS,
+          com.linkedin.kmf.common.Utils.ZK_CONNECTION_TIMEOUT_MS, Integer.MAX_VALUE, Time.SYSTEM, METRIC_GROUP_NAME, "SessionExpireListener", null)) {
         if (!zkClient.reassignPartitionsInProgress()) {
           List<TopicPartitionInfo> partitionInfoList = _adminClient
               .describeTopics(Collections.singleton(_topic)).all().get().get(_topic).partitions();