KAFKA-7388 equal sign in property value for password

core/src/main/scala/kafka/utils/CommandLineUtils.scala

@@ -60,19 +60,15 @@ object CommandLineUtils extends Logging {
    * Parse key-value pairs in the form key=value
    */
   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
   }

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"))
   }

core/src/test/scala/unit/kafka/utils/CommandLineUtilsTest.scala

@@ -29,12 +29,19 @@ class CommandLineUtilsTest {
     CommandLineUtils.parseKeyValueArgs(argArray, false)
   }
 
+  @Test(expected = classOf[java.lang.IllegalArgumentException])
+  def testParseEmptyArgWithNoDelimiter() {
+    val argArray = Array("my.empty.property")
+    CommandLineUtils.parseKeyValueArgs(argArray, 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 with no delimiter should be an empty string",props.getProperty("my.empty.property1"),"")
   }
 
   @Test
@@ -52,4 +59,14 @@ class CommandLineUtilsTest {
     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, 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=")
+    assertEquals("Value of second property should be 'thi=rd'",props.getProperty("third.property"),"thi=rd")
+  }
+
 }