kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject kafka git commit: KAFKA-2641; Upgrade path for ZK authentication
Date Fri, 23 Oct 2015 22:11:16 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 701c46b3a -> 21443f214


KAFKA-2641; Upgrade path for ZK authentication

This pull request adds a configuration parameter and a migration tool. It is also based on
pull request #303, which should go in first.

Author: flavio junqueira <fpj@apache.org>
Author: Flavio Junqueira <fpj@apache.org>
Author: Ismael Juma <ismael@juma.me.uk>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Jun Rao <junrao@gmail.com>

Closes #313 from fpj/KAFKA-2641


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/21443f21
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/21443f21
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/21443f21

Branch: refs/heads/trunk
Commit: 21443f214fc6f1f51037e27f8ece155cf1eb288c
Parents: 701c46b
Author: flavio junqueira <fpj@apache.org>
Authored: Fri Oct 23 15:11:07 2015 -0700
Committer: Jun Rao <junrao@gmail.com>
Committed: Fri Oct 23 15:11:07 2015 -0700

----------------------------------------------------------------------
 bin/zookeeper-security-migration.sh             |  17 ++
 .../apache/kafka/common/security/JaasUtils.java |   5 +
 .../src/main/scala/kafka/admin/AdminUtils.scala |   2 +-
 .../scala/kafka/admin/ZkSecurityMigrator.scala  | 237 +++++++++++++++++++
 .../main/scala/kafka/server/KafkaConfig.scala   |   5 +
 .../main/scala/kafka/server/KafkaServer.scala   |   9 +-
 core/src/main/scala/kafka/utils/ZkUtils.scala   |  15 +-
 .../test/scala/other/kafka/DeleteZKPath.scala   |   1 -
 .../unit/kafka/server/KafkaConfigTest.scala     |   1 +
 9 files changed, 285 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/21443f21/bin/zookeeper-security-migration.sh
----------------------------------------------------------------------
diff --git a/bin/zookeeper-security-migration.sh b/bin/zookeeper-security-migration.sh
new file mode 100755
index 0000000..65fce85
--- /dev/null
+++ b/bin/zookeeper-security-migration.sh
@@ -0,0 +1,17 @@
+#!/bin/bash
+# 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.
+
+exec $(dirname $0)/kafka-run-class.sh kafka.admin.ZkSecurityMigrator $@

http://git-wip-us.apache.org/repos/asf/kafka/blob/21443f21/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java b/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java
index b8c870d..dade986 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java
@@ -109,6 +109,11 @@ public class JaasUtils {
                 throw new KafkaException("Exception while determining if ZooKeeper is secure");
             }
         }
+        /*
+         * Tests fail if we don't reset the login configuration. It is unclear
+         * what is actually triggering this bug.
+         */
+        Configuration.setConfiguration(null);
 
         return isSecurityEnabled;
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/21443f21/core/src/main/scala/kafka/admin/AdminUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AdminUtils.scala b/core/src/main/scala/kafka/admin/AdminUtils.scala
index 6fff176..64527de 100644
--- a/core/src/main/scala/kafka/admin/AdminUtils.scala
+++ b/core/src/main/scala/kafka/admin/AdminUtils.scala
@@ -226,7 +226,7 @@ object AdminUtils extends Logging {
 
   def topicExists(zkUtils: ZkUtils, topic: String): Boolean = 
     zkUtils.zkClient.exists(getTopicPath(topic))
-    
+
   def createTopic(zkUtils: ZkUtils,
                   topic: String,
                   partitions: Int, 

http://git-wip-us.apache.org/repos/asf/kafka/blob/21443f21/core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala b/core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala
new file mode 100644
index 0000000..fce5c03
--- /dev/null
+++ b/core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala
@@ -0,0 +1,237 @@
+/**
+ * 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 kafka.admin
+
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.TimeUnit
+import joptsimple.OptionParser
+import org.I0Itec.zkclient.exception.ZkException
+import kafka.utils.{Logging, ZkUtils, CommandLineUtils}
+import org.apache.log4j.Level
+import org.apache.kafka.common.security.JaasUtils
+import org.apache.zookeeper.AsyncCallback.{ChildrenCallback, StatCallback}
+import org.apache.zookeeper.data.Stat
+import org.apache.zookeeper.KeeperException
+import org.apache.zookeeper.KeeperException.Code
+import scala.annotation.tailrec
+import scala.collection.JavaConverters._
+import scala.collection._
+import scala.collection.mutable.Queue
+import scala.concurrent._
+import scala.concurrent.duration._
+
+/**
+ * This tool is to be used when making access to ZooKeeper authenticated or 
+ * the other way around, when removing authenticated access. The exact steps
+ * to migrate a Kafka cluster from unsecure to secure with respect to ZooKeeper
+ * access are the following:
+ * 
+ * 1- Perform a rolling upgrade of Kafka servers, setting zookeeper.set.acl to false
+ * and passing a valid JAAS login file via the system property 
+ * java.security.auth.login.config
+ * 2- Perform a second rolling upgrade keeping the system property for the login file
+ * and now setting zookeeper.set.acl to true
+ * 3- Finally run this tool. There is a script under ./bin. Run 
+ *   ./bin/zookeeper-security-migration --help
+ * to see the configuration parameters. An example of running it is the following:
+ *  ./bin/zookeeper-security-migration --zookeeper.acl=secure --zookeeper.connection=localhost:2181
+ * 
+ * To convert a cluster from secure to unsecure, we need to perform the following
+ * steps:
+ * 1- Perform a rolling upgrade setting zookeeper.set.acl to false for each server
+ * 2- Run this migration tool, setting zookeeper.acl to unsecure
+ * 3- Perform another rolling upgrade to remove the system property setting the
+ * login file (java.security.auth.login.config).
+ */
+
+object ZkSecurityMigrator extends Logging {
+  val usageMessage = ("ZooKeeper Migration Tool Help. This tool updates the ACLs of "
+                      + "znodes as part of the process of setting up ZooKeeper "
+                      + "authentication.")
+
+  def run(args: Array[String]) {
+    var jaasFile = System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM)
+    val parser = new OptionParser()
+
+    val zkAclOpt = parser.accepts("zookeeper.acl", "Indicates whether to make the Kafka znodes
in ZooKeeper secure or unsecure."
+        + " The options are 'secure' and 'unsecure'").withRequiredArg().ofType(classOf[String])
+    val jaasFileOpt = parser.accepts("jaas.file", "JAAS Config file.").withOptionalArg().ofType(classOf[String])
+    val zkUrlOpt = parser.accepts("zookeeper.connect", "Sets the ZooKeeper connect string
(ensemble). This parameter " +
+      "takes a comma-separated list of host:port pairs.").withRequiredArg().defaultsTo("localhost:2181").
+      ofType(classOf[String])
+    val zkSessionTimeoutOpt = parser.accepts("zookeeper.session.timeout", "Sets the ZooKeeper
session timeout.").
+      withRequiredArg().ofType(classOf[java.lang.Integer]).defaultsTo(30000)
+    val zkConnectionTimeoutOpt = parser.accepts("zookeeper.connection.timeout", "Sets the
ZooKeeper connection timeout.").
+      withRequiredArg().ofType(classOf[java.lang.Integer]).defaultsTo(30000)
+    val helpOpt = parser.accepts("help", "Print usage information.")
+
+    val options = parser.parse(args : _*)
+    if (options.has(helpOpt))
+      CommandLineUtils.printUsageAndDie(parser, usageMessage)
+
+    if ((jaasFile == null) && !options.has(jaasFileOpt)) {
+     val errorMsg = ("No JAAS configuration file has been specified. Please make sure that
you have set either " + 
+                    "the system property %s or the option %s".format(JaasUtils.JAVA_LOGIN_CONFIG_PARAM,
"--jaas.file")) 
+     System.out.println("ERROR: %s".format(errorMsg))
+     throw new IllegalArgumentException("Incorrect configuration")
+    }
+
+    if (jaasFile == null) {
+      jaasFile = options.valueOf(jaasFileOpt)
+      System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, jaasFile)
+    }
+
+    if (!JaasUtils.isZkSecurityEnabled(jaasFile)) {
+      val errorMsg = "Security isn't enabled, most likely the file isn't set properly: %s".format(jaasFile)
+      System.out.println("ERROR: %s".format(errorMsg))
+      throw new IllegalArgumentException("Incorrect configuration") 
+    }
+
+    val zkAcl: Boolean = options.valueOf(zkAclOpt) match {
+      case "secure" =>
+        info("zookeeper.acl option is secure")
+        true
+      case "unsecure" =>
+        info("zookeeper.acl option is unsecure")
+        false
+      case _ =>
+        CommandLineUtils.printUsageAndDie(parser, usageMessage)
+    }
+    val zkUrl = options.valueOf(zkUrlOpt)
+    val zkSessionTimeout = options.valueOf(zkSessionTimeoutOpt).intValue
+    val zkConnectionTimeout = options.valueOf(zkConnectionTimeoutOpt).intValue
+    val zkUtils = ZkUtils(zkUrl, zkSessionTimeout, zkConnectionTimeout, zkAcl)
+    val migrator = new ZkSecurityMigrator(zkUtils)
+    migrator.run()
+  }
+
+  def main(args: Array[String]) {
+    try {
+      run(args)
+    } catch {
+        case e: Exception =>
+          e.printStackTrace()
+    }
+  }
+}
+
+class ZkSecurityMigrator(zkUtils: ZkUtils) extends Logging {
+  private val workQueue = new LinkedBlockingQueue[Runnable]
+  private val futures = new Queue[Future[String]]
+
+  private def setAclsRecursively(path: String) = {
+    info("Setting ACL for path %s".format(path))
+    val setPromise = Promise[String]
+    val childrenPromise = Promise[String]
+    futures.synchronized {
+      futures += setPromise.future
+      futures += childrenPromise.future
+    }
+    zkUtils.zkConnection.getZookeeper.setACL(path, ZkUtils.DefaultAcls(zkUtils.isSecure),
-1, SetACLCallback, setPromise)
+    zkUtils.zkConnection.getZookeeper.getChildren(path, false, GetChildrenCallback, childrenPromise)
+  }
+
+  private object GetChildrenCallback extends ChildrenCallback {
+    def processResult(rc: Int,
+                      path: String,
+                      ctx: Object,
+                      children: java.util.List[String]) {
+      val zkHandle = zkUtils.zkConnection.getZookeeper
+      val promise = ctx.asInstanceOf[Promise[String]]
+      Code.get(rc) match {
+        case Code.OK =>
+          // Set ACL for each child
+          for (child <- children.asScala)
+            setAclsRecursively(s"$path/$child")
+          promise success "done"
+        case Code.CONNECTIONLOSS =>
+          zkHandle.getChildren(path, false, GetChildrenCallback, ctx)
+        case Code.NONODE =>
+          warn("Node is gone, it could be have been legitimately deleted: %s".format(path))
+          promise success "done"
+        case Code.SESSIONEXPIRED =>
+          // Starting a new session isn't really a problem, but it'd complicate
+          // the logic of the tool, so we quit and let the user re-run it.
+          System.out.println("ZooKeeper session expired while changing ACLs")
+          promise failure ZkException.create(KeeperException.create(Code.get(rc)))
+        case _ =>
+          System.out.println("Unexpected return code: %d".format(rc))
+          promise failure ZkException.create(KeeperException.create(Code.get(rc)))
+      }
+    }
+  }
+
+  private object SetACLCallback extends StatCallback {
+    def processResult(rc: Int,
+                      path: String,
+                      ctx: Object,
+                      stat: Stat) {
+      val zkHandle = zkUtils.zkConnection.getZookeeper
+      val promise = ctx.asInstanceOf[Promise[String]]
+
+      Code.get(rc) match {
+        case Code.OK =>
+          info("Successfully set ACLs for %s".format(path))
+          promise success "done"
+        case Code.CONNECTIONLOSS =>
+            zkHandle.setACL(path, ZkUtils.DefaultAcls(zkUtils.isSecure), -1, SetACLCallback,
ctx)
+        case Code.NONODE =>
+          warn("Znode is gone, it could be have been legitimately deleted: %s".format(path))
+          promise success "done"
+        case Code.SESSIONEXPIRED =>
+          // Starting a new session isn't really a problem, but it'd complicate
+          // the logic of the tool, so we quit and let the user re-run it.
+          System.out.println("ZooKeeper session expired while changing ACLs")
+          promise failure ZkException.create(KeeperException.create(Code.get(rc)))
+        case _ =>
+          System.out.println("Unexpected return code: %d".format(rc))
+          promise failure ZkException.create(KeeperException.create(Code.get(rc)))
+      }
+    }
+  }
+
+  private def run(): Unit = {
+    try {
+      for (path <- zkUtils.securePersistentZkPaths) {
+        debug("Going to set ACL for %s".format(path))
+        zkUtils.makeSurePersistentPathExists(path)
+        setAclsRecursively(path)
+      }
+
+      @tailrec
+      def recurse(): Unit = {
+        val future = futures.synchronized { 
+          futures.headOption
+        }
+        future match {
+          case Some(a) =>
+            Await.result(a, 6000 millis)
+            futures.synchronized { futures.dequeue }
+            recurse
+          case None =>
+        }
+      }
+      recurse()
+
+    } finally {
+      zkUtils.close
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/21443f21/core/src/main/scala/kafka/server/KafkaConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 52182b8..d52b5c0 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -39,6 +39,7 @@ object Defaults {
   /** ********* Zookeeper Configuration ***********/
   val ZkSessionTimeoutMs = 6000
   val ZkSyncTimeMs = 2000
+  val ZkEnableSecureAcls = false
 
   /** ********* General Configuration ***********/
   val MaxReservedBrokerId = 1000
@@ -187,6 +188,7 @@ object KafkaConfig {
   val ZkSessionTimeoutMsProp = "zookeeper.session.timeout.ms"
   val ZkConnectionTimeoutMsProp = "zookeeper.connection.timeout.ms"
   val ZkSyncTimeMsProp = "zookeeper.sync.time.ms"
+  val ZkEnableSecureAclsProp = "zookeeper.set.acl"
   /** ********* General Configuration ***********/
   val MaxReservedBrokerIdProp = "reserved.broker.max.id"
   val BrokerIdProp = "broker.id"
@@ -334,6 +336,7 @@ object KafkaConfig {
   val ZkSessionTimeoutMsDoc = "Zookeeper session timeout"
   val ZkConnectionTimeoutMsDoc = "The max time that the client waits to establish a connection
to zookeeper"
   val ZkSyncTimeMsDoc = "How far a ZK follower can be behind a ZK leader"
+  val ZkEnableSecureAclsDoc = "Set client to use secure ACLs"
   /** ********* General Configuration ***********/
   val MaxReservedBrokerIdDoc = "Max number that can be used for a broker.id"
   val BrokerIdDoc = "The broker id for this server. " +
@@ -515,6 +518,7 @@ object KafkaConfig {
       .define(ZkSessionTimeoutMsProp, INT, Defaults.ZkSessionTimeoutMs, HIGH, ZkSessionTimeoutMsDoc)
       .define(ZkConnectionTimeoutMsProp, INT, HIGH, ZkConnectionTimeoutMsDoc, false)
       .define(ZkSyncTimeMsProp, INT, Defaults.ZkSyncTimeMs, LOW, ZkSyncTimeMsDoc)
+      .define(ZkEnableSecureAclsProp, BOOLEAN, Defaults.ZkEnableSecureAcls, HIGH, ZkEnableSecureAclsDoc)
 
       /** ********* General Configuration ***********/
       .define(MaxReservedBrokerIdProp, INT, Defaults.MaxReservedBrokerId, atLeast(0), MEDIUM,
MaxReservedBrokerIdProp)
@@ -702,6 +706,7 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka
   val zkConnectionTimeoutMs: Int =
     Option(getInt(KafkaConfig.ZkConnectionTimeoutMsProp)).map(_.toInt).getOrElse(getInt(KafkaConfig.ZkSessionTimeoutMsProp))
   val zkSyncTimeMs: Int = getInt(KafkaConfig.ZkSyncTimeMsProp)
+  val zkEnableSecureAcls: Boolean = getBoolean(KafkaConfig.ZkEnableSecureAclsProp)
 
   /** ********* General Configuration ***********/
   val maxReservedBrokerId: Int = getInt(KafkaConfig.MaxReservedBrokerIdProp)

http://git-wip-us.apache.org/repos/asf/kafka/blob/21443f21/core/src/main/scala/kafka/server/KafkaServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index 617f807..c6ea84e 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -260,12 +260,17 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime,
threadNamePr
         ""
     }
 
+    val secureAclsEnabled = JaasUtils.isZkSecurityEnabled(System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM))
&& config.zkEnableSecureAcls
+    
+    if(config.zkEnableSecureAcls && !secureAclsEnabled) {
+      throw new java.lang.SecurityException("zkEnableSecureAcls is true, but the verification
of the JAAS login file failed.")    
+    }
     if (chroot.length > 1) {
       val zkConnForChrootCreation = config.zkConnect.substring(0, config.zkConnect.indexOf("/"))
       val zkClientForChrootCreation = ZkUtils(zkConnForChrootCreation, 
                                               config.zkSessionTimeoutMs,
                                               config.zkConnectionTimeoutMs,
-                                              JaasUtils.isZkSecurityEnabled(System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM)))
+                                              secureAclsEnabled)
       zkClientForChrootCreation.makeSurePersistentPathExists(chroot)
       info("Created zookeeper path " + chroot)
       zkClientForChrootCreation.zkClient.close()
@@ -274,7 +279,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr
     val zkUtils = ZkUtils(config.zkConnect,
                           config.zkSessionTimeoutMs,
                           config.zkConnectionTimeoutMs,
-                          JaasUtils.isZkSecurityEnabled(System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM)))
+                          secureAclsEnabled)
     zkUtils.setupCommonPaths()
     zkUtils
   }

http://git-wip-us.apache.org/repos/asf/kafka/blob/21443f21/core/src/main/scala/kafka/utils/ZkUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala
index 17e63e2..a39e61c 100644
--- a/core/src/main/scala/kafka/utils/ZkUtils.scala
+++ b/core/src/main/scala/kafka/utils/ZkUtils.scala
@@ -112,7 +112,7 @@ object ZkUtils {
    * Get calls that only depend on static paths
    */
   def getTopicPath(topic: String): String = {
-    BrokerTopicsPath + "/" + topic
+    ZkUtils.BrokerTopicsPath + "/" + topic
   }
 
   def getTopicPartitionsPath(topic: String): String = {
@@ -126,7 +126,7 @@ object ZkUtils {
     getTopicPartitionPath(topic, partitionId) + "/" + "state"
     
   def getEntityConfigRootPath(entityType: String): String =
-    EntityConfigPath + "/" + entityType
+    ZkUtils.EntityConfigPath + "/" + entityType
 
   def getEntityConfigPath(entityType: String, entity: String): String =
     getEntityConfigRootPath(entityType) + "/" + entity
@@ -149,6 +149,15 @@ class ZkUtils(val zkClient: ZkClient,
                               BrokerSequenceIdPath,
                               IsrChangeNotificationPath)
 
+  val securePersistentZkPaths = Seq(BrokerIdsPath,
+                                    BrokerTopicsPath,
+                                    EntityConfigChangesPath,
+                                    getEntityConfigRootPath(ConfigType.Topic),
+                                    getEntityConfigRootPath(ConfigType.Client),
+                                    DeleteTopicsPath,
+                                    BrokerSequenceIdPath,
+                                    IsrChangeNotificationPath)
+
   val DefaultAcls: java.util.List[ACL] = ZkUtils.DefaultAcls(isSecure)
   
   def getController(): Int = {
@@ -713,7 +722,7 @@ class ZkUtils(val zkClient: ZkClient,
   def deletePartition(brokerId: Int, topic: String) {
     val brokerIdPath = BrokerIdsPath + "/" + brokerId
     zkClient.delete(brokerIdPath)
-    val brokerPartTopicPath = BrokerTopicsPath + "/" + topic + "/" + brokerId
+    val brokerPartTopicPath = ZkUtils.BrokerTopicsPath + "/" + topic + "/" + brokerId
     zkClient.delete(brokerPartTopicPath)
   }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/21443f21/core/src/test/scala/other/kafka/DeleteZKPath.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/other/kafka/DeleteZKPath.scala b/core/src/test/scala/other/kafka/DeleteZKPath.scala
index 92bde88..202bf43 100755
--- a/core/src/test/scala/other/kafka/DeleteZKPath.scala
+++ b/core/src/test/scala/other/kafka/DeleteZKPath.scala
@@ -31,7 +31,6 @@ object DeleteZKPath {
 
     val config = new ConsumerConfig(Utils.loadProps(args(0)))
     val zkPath = args(1)
-
     val zkUtils = ZkUtils(config.zkConnect, config.zkSessionTimeoutMs, config.zkConnectionTimeoutMs,
false)
 
     try {

http://git-wip-us.apache.org/repos/asf/kafka/blob/21443f21/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index b1a7f21..4059dc2 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -398,6 +398,7 @@ class KafkaConfigTest {
         case KafkaConfig.ZkSessionTimeoutMsProp => assertPropertyInvalid(getBaseProperties(),
name, "not_a_number")
         case KafkaConfig.ZkConnectionTimeoutMsProp => assertPropertyInvalid(getBaseProperties(),
name, "not_a_number")
         case KafkaConfig.ZkSyncTimeMsProp => assertPropertyInvalid(getBaseProperties(),
name, "not_a_number")
+        case KafkaConfig.ZkEnableSecureAclsProp => assertPropertyInvalid(getBaseProperties(),
name, "not_a_boolean")
 
         case KafkaConfig.BrokerIdProp => assertPropertyInvalid(getBaseProperties(), name,
"not_a_number")
         case KafkaConfig.NumNetworkThreadsProp => assertPropertyInvalid(getBaseProperties(),
name, "not_a_number", "0")


Mime
View raw message