MINOR: Rejoin split ssl principal mapping rules (#6099)

* Join ssl principal mapping rules correctly before evaluating.

Java properties splits the configuration array on commas, and that leads to rules containing commas being split before being evaluated. This commit adds a code change to re-join those strings into full rules before evaluating them. The function assumes every rule is either DEFAULT or begins with the prefix RULE:

clients/src/main/java/org/apache/kafka/common/security/ssl/SslPrincipalMapper.java

@@ -17,9 +17,9 @@
 package org.apache.kafka.common.security.ssl;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
+import java.util.Collections;
+import java.util.ArrayList;
 import java.util.Locale;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -39,7 +39,32 @@ public class SslPrincipalMapper {
         return new SslPrincipalMapper(parseRules(rules));
     }
 
+    private static List<String> joinSplitRules(List<String> rules) {
+        String rule = "RULE:";
+        String defaultRule = "DEFAULT";
+        List<String> retVal = new ArrayList<>();
+        StringBuilder currentRule = new StringBuilder();
+        for (String r : rules) {
+            if (currentRule.length() > 0) {
+                if (r.startsWith(rule) || r.equals(defaultRule)) {
+                    retVal.add(currentRule.toString());
+                    currentRule.setLength(0);
+                    currentRule.append(r);
+                } else {
+                    currentRule.append(String.format(",%s", r));
+                }
+            } else {
+                currentRule.append(r);
+            }
+        }
+        if (currentRule.length() > 0) {
+            retVal.add(currentRule.toString());
+        }
+        return retVal;
+    }
+
     private static List<Rule> parseRules(List<String> rules) {
+        rules = joinSplitRules(rules);
         List<Rule> result = new ArrayList<>();
         for (String rule : rules) {
             Matcher matcher = RULE_PARSER.matcher(rule);

clients/src/test/java/org/apache/kafka/common/security/ssl/SslPrincipalMapperTest.java

@@ -36,6 +36,16 @@ public class SslPrincipalMapperTest {
         testValidRule(Arrays.asList("RULE:^cn=(.?),ou=(.?),dc=(.?),dc=(.?)$/$1@$2/U"));
     }
 
+    @Test
+    public void testValidSplitRules() {
+        testValidRule(Arrays.asList("DEFAULT"));
+        testValidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/$1/"));
+        testValidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/$1/L", "DEFAULT"));
+        testValidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=(.*?),O=(.*?),L=(.*?)", "ST=(.*?)", "C=(.*?)$/$1@$2/"));
+        testValidRule(Arrays.asList("RULE:^.*[Cc][Nn]=([a-zA-Z0-9.]*).*$/$1/L"));
+        testValidRule(Arrays.asList("RULE:^cn=(.?)", "ou=(.?)", "dc=(.?)", "dc=(.?)$/$1@$2/U"));
+    }
+
     private void testValidRule(List<String> rules) {
         SslPrincipalMapper.fromRules(rules);
     }
@@ -55,6 +65,21 @@ public class SslPrincipalMapperTest {
         testInvalidRule(Arrays.asList("RULE:^CN=(.*?),OU=ServiceUsers.*$/LU"));
     }
 
+    @Test
+    public void testInvalidSplitRules() {
+        testInvalidRule(Arrays.asList("default"));
+        testInvalidRule(Arrays.asList("DEFAUL"));
+        testInvalidRule(Arrays.asList("DEFAULT/L"));
+        testInvalidRule(Arrays.asList("DEFAULT/U"));
+
+        testInvalidRule(Arrays.asList("RULE:CN=(.*?)", "OU=ServiceUsers.*/$1"));
+        testInvalidRule(Arrays.asList("rule:^CN=(.*?)", "OU=ServiceUsers.*$/$1/"));
+        testInvalidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/$1/L/U"));
+        testInvalidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/L"));
+        testInvalidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/U"));
+        testInvalidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/LU"));
+    }
+
     private void testInvalidRule(List<String> rules) {
         try {
             System.out.println(SslPrincipalMapper.fromRules(rules));