Index: core/src/main/scala/kafka/utils/CommandLineUtils.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- core/src/main/scala/kafka/utils/CommandLineUtils.scala	(revision b1a1cae880817332f172dccaa5e3e4a69228074f)
+++ core/src/main/scala/kafka/utils/CommandLineUtils.scala	(revision )
@@ -60,14 +60,17 @@
    * Parse key-value pairs in the form key=value
    */
   def parseKeyValueArgs(args: Iterable[String]): Properties = {
-    val splits = args.map(_ split "=").filterNot(_ == null).filterNot(_.length == 0)
-    if(!splits.forall(_.length == 2)) {
+    val splits = args.map(_ split "=").filterNot(_.length == 0)
+
+    val props = new Properties
+    for(a <- splits) {
+      if (a.length == 1) props.put(a(0), "")
+      else if (a.length == 2) props.put(a(0), a(1))
+      else {
-      System.err.println("Invalid command line properties: " + args.mkString(" "))
-      System.exit(1)
-    }
+        System.err.println("Invalid command line properties: " + args.mkString(" "))
+        System.exit(1)
+      }
-    val props = new Properties
-    for(a <- splits)
-      props.put(a(0), a(1))
+    }
     props
   }
 }
\ No newline at end of file
Index: core/src/test/scala/unit/kafka/utils/CommandLineUtilsTest.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- core/src/test/scala/unit/kafka/utils/CommandLineUtilsTest.scala	(revision )
+++ core/src/test/scala/unit/kafka/utils/CommandLineUtilsTest.scala	(revision )
@@ -0,0 +1,49 @@
+/**
+ * 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 unit.kafka.utils
+
+import junit.framework.Assert._
+import org.junit.{Test, After, Before}
+import kafka.utils.CommandLineUtils;
+
+class CommandLineUtilsTest {
+
+
+  @Test
+  def testParseEmptyArg() {
+    val argArray = Array("my.empty.property=")
+    val props = CommandLineUtils.parseKeyValueArgs(argArray)
+    assertEquals("Empty value should be equal to empty string",props.getProperty("my.empty.property"),"")
+  }
+
+  @Test
+  def testParseSingleArg() {
+    val argArray = Array("my.property=value")
+    val props = CommandLineUtils.parseKeyValueArgs(argArray)
+    assertEquals("Value of a single property should be 'value' ",props.getProperty("my.property"),"value")
+  }
+
+  @Test
+  def testParseArgs() {
+    val argArray = Array("first.property=first","second.property=second")
+    val props = CommandLineUtils.parseKeyValueArgs(argArray)
+    assertEquals("Value of first property should be 'first'",props.getProperty("first.property"),"first")
+    assertEquals("Value of second property should be 'second'",props.getProperty("second.property"),"second")
+  }
+
+}