[fix][client] Fix the Windows absolute path not recognized in auth param string

pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java

@@ -27,7 +27,6 @@
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.when;
 import com.google.common.collect.Sets;
-import com.google.common.io.Resources;
 import com.google.common.util.concurrent.MoreExecutors;
 import io.netty.channel.EventLoopGroup;
 import java.lang.reflect.Field;
@@ -79,6 +78,7 @@
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 import org.apache.pulsar.metadata.impl.ZKMetadataStore;
 import org.apache.pulsar.tests.TestRetrySupport;
+import org.apache.pulsar.utils.ResourceUtils;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.MockZooKeeper;
 import org.apache.zookeeper.MockZooKeeperSession;
@@ -93,20 +93,20 @@
  */
 public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
     public final static String BROKER_KEYSTORE_FILE_PATH =
-            Resources.getResource("certificate-authority/jks/broker.keystore.jks").getPath();
+            ResourceUtils.getAbsolutePath("certificate-authority/jks/broker.keystore.jks");
     public final static String BROKER_TRUSTSTORE_FILE_PATH =
-            Resources.getResource("certificate-authority/jks/broker.truststore.jks").getPath();
+            ResourceUtils.getAbsolutePath("certificate-authority/jks/broker.truststore.jks");
     public final static String BROKER_TRUSTSTORE_NO_PASSWORD_FILE_PATH =
-            Resources.getResource("certificate-authority/jks/broker.truststore.nopassword.jks").getPath();
+            ResourceUtils.getAbsolutePath("certificate-authority/jks/broker.truststore.nopassword.jks");
     public final static String BROKER_KEYSTORE_PW = "111111";
     public final static String BROKER_TRUSTSTORE_PW = "111111";
 
     public final static String CLIENT_KEYSTORE_FILE_PATH =
-            Resources.getResource("certificate-authority/jks/client.keystore.jks").getPath();
+            ResourceUtils.getAbsolutePath("certificate-authority/jks/client.keystore.jks");
     public final static String CLIENT_TRUSTSTORE_FILE_PATH =
-            Resources.getResource("certificate-authority/jks/client.truststore.jks").getPath();
+            ResourceUtils.getAbsolutePath("certificate-authority/jks/client.truststore.jks");
     public final static String CLIENT_TRUSTSTORE_NO_PASSWORD_FILE_PATH =
-            Resources.getResource("certificate-authority/jks/client.truststore.nopassword.jks").getPath();
+            ResourceUtils.getAbsolutePath("certificate-authority/jks/client.truststore.nopassword.jks");
     public final static String CLIENT_KEYSTORE_PW = "111111";
     public final static String CLIENT_TRUSTSTORE_PW = "111111";
 

pulsar-broker/src/test/java/org/apache/pulsar/utils/ResourceUtils.java

@@ -0,0 +1,31 @@
+/*
+ * 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 org.apache.pulsar.utils;
+
+import com.google.common.io.Resources;
+import java.io.File;
+
+public class ResourceUtils {
+
+    public static String getAbsolutePath(String resourceName) {
+        // On Windows, URL#getPath might return a string that starts with a disk name, e.g. "/C:/"
+        // It's invalid to use this path to open a file, so we need to get the absolute path via File.
+        return new File(Resources.getResource(resourceName).getPath()).getAbsolutePath();
+    }
+}

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AuthenticationUtil.java

@@ -43,9 +43,12 @@ public static Map<String, String> configureFromPulsar1AuthParamString(String aut
         if (isNotBlank(authParamsString)) {
             String[] params = authParamsString.split(",");
             for (String p : params) {
-                String[] kv = p.split(":");
-                if (kv.length == 2) {
-                    authParams.put(kv[0], kv[1]);
+                // The value could be a file path, which could contain a colon like "C:\\path\\to\\file" on Windows.
+                int index = p.indexOf(':');
+                String key = p.substring(0, index);
+                String value = p.substring(index + 1);
+                if (!key.isEmpty() && !value.isEmpty()) {
+                    authParams.put(key, value);
                 }
             }
         }

pulsar-client/src/test/java/org/apache/pulsar/client/impl/AuthenticationUtilTest.java

@@ -0,0 +1,37 @@
+/*
+ * 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 org.apache.pulsar.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import java.util.Map;
+import org.testng.annotations.Test;
+
+public class AuthenticationUtilTest {
+
+    @Test
+    public void testConfigureAuthParamString() {
+        Map<String, String> params = AuthenticationUtil.configureFromPulsar1AuthParamString(
+                "key:value,path:C:\\path\\to\\file,null-key:,:null-value,:,key:value-2");
+        assertEquals(params.size(), 2);
+        assertEquals(params.get("key"), "value-2");
+        assertEquals(params.get("path"), "C:\\path\\to\\file");
+        assertFalse(params.containsKey("null-key"));
+    }
+}