updated to kafka 1.1.0 (#321)

* updated to kafka 1.1.0

* fixed for kafka 1.1.0

* fixed for kafka 1.1.0

* fixed for kafka 1.1.0 - empty commit

* Revert "fixed for kafka 1.1.0 - empty commit"

This reverts commit e5fc7fe

Author: jacum

.travis.yml

@@ -20,14 +20,15 @@ jdk:
   - oraclejdk8
   - oraclejdk7
 env:
+  - KAFKA_VER=1.1.0
   - KAFKA_VER=0.11.0.0
   - KAFKA_VER=0.10.2.0
   - KAFKA_VER=0.9.0.1
   - KAFKA_VER=0.8.2.2
 matrix:
   exclude:
   - jdk: oraclejdk7
-    env: KAFKA_VER=0.11.0.0
+    env: KAFKA_VER=1.1.0
   - jdk: oraclejdk7
     env: KAFKA_VER=0.9.0.1
   - jdk: oraclejdk7

build.gradle

@@ -8,12 +8,12 @@ plugins {
 
 def scalaVersion = "2.11.8"
 def mesosVersion = "0.28.0"
-ext.kafkaVersion = project.hasProperty("kafkaVersion") ? project.getProperty("kafkaVersion") : "0.11.0.0"
+ext.kafkaVersion = project.hasProperty("kafkaVersion") ? project.getProperty("kafkaVersion") : "1.1.0"
 def kafkaMajorVersion="${kafkaVersion.split("\\.").take(2).join("_")}"
 def kafkaScalaVersion=scalaVersion.split("\\.").take(2).join(".")
 def jettyVersion = "9.0.4.v20130625"
 def jerseyVersion = "2.24"
-def jacksonVersion = "2.8.2"
+def jacksonVersion = "2.9.5"
 
 version = new File('src/scala/main/ly/stealth/mesos/kafka/Config.scala')
         .readLines()

gradle/wrapper/gradle-wrapper.properties

@@ -3,4 +3,4 @@ distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.7-bin.zip

src/docker/Dockerfile

@@ -1,26 +1,4 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-FROM ubuntu:14.04
-
-RUN \
-  if [ -f "/tmp/apt-proxy" ]; then \
-    APT_PROXY=$(cat "/tmp/apt-proxy"); \
-    echo "Using apt-proxy: $APT_PROXY"; \
-    echo "Acquire::http::Proxy \"$APT_PROXY\";" > /etc/apt/apt.conf.d/90-apt-proxy.conf ; \
-  fi
+FROM ubuntu:16.04
 
 # Add mesos repo
 RUN \
@@ -31,8 +9,8 @@ RUN \
 RUN \
   apt-get update && \
   apt-get install -qy \
-    git vim zip mc curl \
-    openjdk-7-jdk mesos libc6
+    git zip mc curl \
+    openjdk-8-jdk mesos libc6
 
 # Add kafka-mesos & kafka
 COPY .docker/kafka* /opt/kafka-mesos/

src/docker/build-image.sh

@@ -5,8 +5,8 @@ base_dir=../..
 tmp_dir=.docker
 mkdir -p $tmp_dir
 
-kafka_version=0.8.2.2
-scala_version=2.10
+kafka_version=1.1.0
+scala_version=2.11
 docker_tag=`whoami`/kafka-mesos
 
 print_help() {

src/scala/iface/1_1/ly/stealth/mesos/kafka/interface/impl/AdminUtils.scala

@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ly.stealth.mesos.kafka.interface.impl
+
+import kafka.utils.{ZkUtils => KafkaZkUtils}
+import kafka.admin.{BrokerMetadata, AdminUtils => KafkaAdminUtils}
+import kafka.zk.AdminZkClient
+import java.util.Properties
+import ly.stealth.mesos.kafka.interface.{AdminUtilsProxy, FeatureSupport}
+import scala.collection.Map
+
+
+class AdminUtils(zkUrl: String) extends AdminUtilsProxy {
+  private val DEFAULT_TIMEOUT_MS = 30000
+  private val zkUtils = KafkaZkUtils(zkUrl, DEFAULT_TIMEOUT_MS, DEFAULT_TIMEOUT_MS, isZkSecurityEnabled = false)
+
+  override def fetchAllTopicConfigs(): Map[String, Properties] = KafkaAdminUtils.fetchAllTopicConfigs(zkUtils)
+
+  override def createOrUpdateTopicPartitionAssignmentPathInZK(
+    topic: String,
+    partitionReplicaAssignment: Map[Int, Seq[Int]],
+    config: Properties,
+    update: Boolean
+  ): Unit = KafkaAdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, topic, partitionReplicaAssignment, config, update)
+
+  override def changeTopicConfig(
+    topic: String,
+    configs: Properties
+  ): Unit = KafkaAdminUtils.changeTopicConfig(zkUtils, topic, configs)
+
+  override def deleteTopic(topicToDelete: String): Unit =
+    KafkaAdminUtils.deleteTopic(zkUtils, topicToDelete)
+
+  override def fetchEntityConfig(
+    entityType: String,
+    entity: String
+  ): Properties = KafkaAdminUtils.fetchEntityConfig(zkUtils, entityType, entity)
+
+  override def changeClientIdConfig(
+    clientId: String,
+    configs: Properties
+  ): Unit = KafkaAdminUtils.changeClientIdConfig(zkUtils, clientId, configs)
+
+  override def fetchAllEntityConfigs(entityType: String): Map[String, Properties]
+  = KafkaAdminUtils.fetchAllEntityConfigs(zkUtils, entityType)
+
+  override def assignReplicasToBrokers(
+    ids: Seq[Int],
+    nPartitions: Int,
+    replicationFactor: Int,
+    fixedStartIndex: Int,
+    startPartitionId: Int
+  ): Map[Int, Seq[Int]] = {
+    val md = ids.map(BrokerMetadata(_, None))
+    KafkaAdminUtils.assignReplicasToBrokers(md, nPartitions, replicationFactor, fixedStartIndex, startPartitionId)
+  }
+
+  override val features: FeatureSupport = FeatureSupport(quotas = true, genericEntityConfigs = true)
+}

src/scala/iface/1_1/ly/stealth/mesos/kafka/interface/impl/ZkUtils.scala

@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ly.stealth.mesos.kafka.interface.impl
+
+import kafka.utils.{ZkUtils => KafkaZkUtils}
+import kafka.common.TopicAndPartition
+import kafka.controller.{LeaderIsrAndControllerEpoch, ReassignedPartitionsContext}
+import ly.stealth.mesos.kafka.interface.ZkUtilsProxy
+import scala.collection.{Map, Set, mutable}
+
+class ZkUtils(zkUrl: String) extends ZkUtilsProxy {
+  private val DEFAULT_TIMEOUT_MS = 30000
+  private val zkUtils = KafkaZkUtils(zkUrl, DEFAULT_TIMEOUT_MS, DEFAULT_TIMEOUT_MS, isZkSecurityEnabled = false)
+
+  override def getAllTopics(): Seq[String] = zkUtils.getAllTopics()
+
+  override def getReplicaAssignmentForTopics(topics: Seq[String]): mutable.Map[TopicAndPartition, Seq[Int]]
+    = zkUtils.getReplicaAssignmentForTopics(topics)
+
+  override def getPartitionsBeingReassigned(): Map[TopicAndPartition, ReassignedPartitionsContext]
+    = zkUtils.getPartitionsBeingReassigned()
+
+  override def getReplicasForPartition(
+    topic: String,
+    partition: Int
+  ): Seq[Int] = zkUtils.getReplicasForPartition(topic, partition)
+
+  override def updatePartitionReassignmentData(partitionsToBeReassigned: Map[TopicAndPartition, Seq[Int]]): Unit
+    = zkUtils.updatePartitionReassignmentData(partitionsToBeReassigned)
+
+  override def createPersistentPath(
+    path: String,
+    data: String
+  ): Unit = zkUtils.createPersistentPath(path, data)
+
+  override def getPartitionAssignmentForTopics(topics: Seq[String]): mutable.Map[String, Map[Int, Seq[Int]]]
+    = zkUtils.getPartitionAssignmentForTopics(topics)
+
+  override def getPartitionLeaderAndIsrForTopics(topicAndPartitions: Set[TopicAndPartition]): mutable.Map[TopicAndPartition, LeaderIsrAndControllerEpoch]
+    = zkUtils.getPartitionLeaderAndIsrForTopics(topicAndPartitions)
+
+  override def getSortedBrokerList(): Seq[Int] = zkUtils.getSortedBrokerList()
+}