diff --git a/core/src/main/scala/kafka/utils/CommandLineUtils.scala b/core/src/main/scala/kafka/utils/CommandLineUtils.scala
index edf473e3a708..700a137c8def 100644
--- a/core/src/main/scala/kafka/utils/CommandLineUtils.scala
+++ b/core/src/main/scala/kafka/utils/CommandLineUtils.scala
@@ -5,7 +5,7 @@
  * 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
@@ -58,21 +58,18 @@ object CommandLineUtils extends Logging {
 
   /**
    * Parse key-value pairs in the form key=value
+   * value may contain equals sign
    */
   def parseKeyValueArgs(args: Iterable[String], acceptMissingValue: Boolean = true): Properties = {
-    val splits = args.map(_ split "=").filterNot(_.length == 0)
+    val splits = args.map(_.split("=", 2)).filterNot(_.length == 0)
 
     val props = new Properties
     for (a <- splits) {
-      if (a.length == 1) {
+      if (a.length == 1 || (a.length == 2 && a(1).isEmpty())) {
         if (acceptMissingValue) props.put(a(0), "")
         else throw new IllegalArgumentException(s"Missing value for key ${a(0)}")
       }
-      else if (a.length == 2) props.put(a(0), a(1))
-      else {
-        System.err.println("Invalid command line properties: " + args.mkString(" "))
-        Exit.exit(1)
-      }
+      else props.put(a(0), a(1))
     }
     props
   }
diff --git a/core/src/test/scala/unit/kafka/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/KafkaConfigTest.scala
index 469cf92377be..c8876327dd2d 100644
--- a/core/src/test/scala/unit/kafka/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/KafkaConfigTest.scala
@@ -58,12 +58,6 @@ class KafkaTest {
     assertEquals(util.Arrays.asList("compact","delete"), config4.logCleanupPolicy)
   }
 
-  @Test(expected = classOf[FatalExitError])
-  def testGetKafkaConfigFromArgsWrongSetValue(): Unit = {
-    val propertiesFile = prepareDefaultConfig()
-    KafkaConfig.fromProps(Kafka.getPropsFromArgs(Array(propertiesFile, "--override", "a=b=c")))
-  }
-
   @Test(expected = classOf[FatalExitError])
   def testGetKafkaConfigFromArgsNonArgsAtTheEnd(): Unit = {
     val propertiesFile = prepareDefaultConfig()
@@ -97,6 +91,23 @@ class KafkaTest {
     assertEquals("truststore_password", config.getPassword(KafkaConfig.SslTruststorePasswordProp).value)
   }
 
+  @Test
+  def testKafkaSslPasswordsWithSymbols(): Unit = {
+    val password = "=!#-+!?*/\"\'^%$=\\.,@:;="
+    val propertiesFile = prepareDefaultConfig()
+    val config = KafkaConfig.fromProps(Kafka.getPropsFromArgs(Array(propertiesFile,
+      "--override", "ssl.keystore.password=" + password,
+      "--override", "ssl.key.password=" + password,
+      "--override", "ssl.truststore.password=" + password)))
+    assertEquals(Password.HIDDEN, config.getPassword(KafkaConfig.SslKeyPasswordProp).toString)
+    assertEquals(Password.HIDDEN, config.getPassword(KafkaConfig.SslKeystorePasswordProp).toString)
+    assertEquals(Password.HIDDEN, config.getPassword(KafkaConfig.SslTruststorePasswordProp).toString)
+
+    assertEquals(password, config.getPassword(KafkaConfig.SslKeystorePasswordProp).value)
+    assertEquals(password, config.getPassword(KafkaConfig.SslKeyPasswordProp).value)
+    assertEquals(password, config.getPassword(KafkaConfig.SslTruststorePasswordProp).value)
+  }
+
   def prepareDefaultConfig(): String = {
     prepareConfig(Array("broker.id=1", "zookeeper.connect=somewhere"))
   }
diff --git a/core/src/test/scala/unit/kafka/utils/CommandLineUtilsTest.scala b/core/src/test/scala/unit/kafka/utils/CommandLineUtilsTest.scala
index 50023f804640..25c67291210d 100644
--- a/core/src/test/scala/unit/kafka/utils/CommandLineUtilsTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/CommandLineUtilsTest.scala
@@ -26,30 +26,51 @@ class CommandLineUtilsTest {
   @Test(expected = classOf[java.lang.IllegalArgumentException])
   def testParseEmptyArg() {
     val argArray = Array("my.empty.property=")
-    CommandLineUtils.parseKeyValueArgs(argArray, false)
+
+    CommandLineUtils.parseKeyValueArgs(argArray, acceptMissingValue = false)
   }
 
+  @Test(expected = classOf[java.lang.IllegalArgumentException])
+  def testParseEmptyArgWithNoDelimiter() {
+    val argArray = Array("my.empty.property")
+
+    CommandLineUtils.parseKeyValueArgs(argArray, acceptMissingValue = false)
+  }
 
   @Test
   def testParseEmptyArgAsValid() {
-    val argArray = Array("my.empty.property=")
+    val argArray = Array("my.empty.property=", "my.empty.property1")
     val props = CommandLineUtils.parseKeyValueArgs(argArray)
-    assertEquals("Value of a key with missing value should be an empty string",props.getProperty("my.empty.property"),"")
+
+    assertEquals("Value of a key with missing value should be an empty string", props.getProperty("my.empty.property"), "")
+    assertEquals("Value of a key with missing value with no delimiter should be an empty string", props.getProperty("my.empty.property1"), "")
   }
 
   @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")
+
+    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, false)
-    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")
+    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")
+  }
+
+  @Test
+  def testParseArgsWithMultipleDelimiters() {
+    val argArray = Array("first.property==first", "second.property=second=", "third.property=thi=rd")
+    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=")
+    assertEquals("Value of second property should be 'thi=rd'", props.getProperty("third.property"), "thi=rd")
   }
 
 }