Parse authParamsString on plugin side (#671) (#721)

pulsar-client/src/main/java/org/apache/pulsar/client/api/Authentication.java

@@ -43,7 +43,9 @@ public interface Authentication extends Closeable, Serializable {
      * Configure the authentication plugins with the supplied parameters
      *
      * @param authParams
+     * @deprecated This method will be deleted on versin 2.0, instead please use configure(String encodedAuthParamString) which is in EncodedAuthenticationParameterSupport for now and will be integrated into this interface.
      */
+    @Deprecated
     void configure(Map<String, String> authParams);
 
     /**

pulsar-client/src/main/java/org/apache/pulsar/client/api/AuthenticationFactory.java

@@ -19,48 +19,65 @@
 package org.apache.pulsar.client.api;
 
 import java.util.Map;
+import java.util.HashMap;
 
-import org.apache.pulsar.client.api.Authentication;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
 import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
 import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
-
-import java.util.HashMap;
-import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import org.apache.pulsar.client.api.EncodedAuthenticationParameterSupport;
 
 public final class AuthenticationFactory {
 
+    private static Map<String, String> parseAuthParamsString(String authParamsString) {
+        Map<String, String> authParams = new HashMap<>();
+
+        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]);
+                }
+            }
+        }
+        return authParams;
+    }
+
     /**
      * Create an instance of the Authentication-Plugin
      *
-     * @param authPluginClassName
-     *            name of the Authentication-Plugin you want to use
-     * @param authParamsString
-     *            string which represents parameters for the Authentication-Plugin, e.g., "key1:val1,key2:val2"
+     * @param authPluginClassName name of the Authentication-Plugin you want to use
+     * @param authParamsString    string which represents parameters for the Authentication-Plugin, e.g., "key1:val1,key2:val2"
      * @return instance of the Authentication-Plugin
      * @throws UnsupportedAuthenticationException
      */
     public static final Authentication create(String authPluginClassName, String authParamsString)
             throws UnsupportedAuthenticationException {
-        Map<String, String> authParams = new HashMap<String, String>();
-        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]);
+        try {
+            if (isNotBlank(authPluginClassName)) {
+                Class<?> authClass = Class.forName(authPluginClassName);
+                Authentication auth = (Authentication) authClass.newInstance();
+                if (auth instanceof EncodedAuthenticationParameterSupport) {
+                    // Parse parameters on plugin side.
+                    ((EncodedAuthenticationParameterSupport) auth).configure(authParamsString);
+                } else {
+                    // Parse parameters by default parse logic.
+                    auth.configure(parseAuthParamsString(authParamsString));
                 }
+                return auth;
+            } else {
+                return new AuthenticationDisabled();
             }
+        } catch (Throwable t) {
+            throw new UnsupportedAuthenticationException(t);
         }
-        return AuthenticationFactory.create(authPluginClassName, authParams);
     }
 
     /**
      * Create an instance of the Authentication-Plugin
      *
-     * @param authPluginClassName
-     *            name of the Authentication-Plugin you want to use
-     * @param authParams
-     *            map which represents parameters for the Authentication-Plugin
+     * @param authPluginClassName name of the Authentication-Plugin you want to use
+     * @param authParams          map which represents parameters for the Authentication-Plugin
      * @return instance of the Authentication-Plugin
      * @throws UnsupportedAuthenticationException
      */

pulsar-client/src/main/java/org/apache/pulsar/client/api/EncodedAuthenticationParameterSupport.java

@@ -0,0 +1,30 @@
+/**
+ * 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.api;
+
+public interface EncodedAuthenticationParameterSupport {
+
+    /**
+     * Plugins which use ":" and/or "," in a configuration parameter value need to implement this interface.
+     * This interface will be integrated into Authentication interface and be required for all plugins on version 2.0.
+     *
+     * @param encodedAuthParamString
+     */
+    void configure(String encodedAuthParamString);
+}

pulsar-client/src/test/java/org/apache/pulsar/client/api/AuthenticationTest.java

@@ -0,0 +1,85 @@
+/**
+ * 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.api;
+
+import org.apache.pulsar.client.impl.auth.MockEncodedAuthenticationParameterSupport;
+import org.apache.pulsar.client.impl.auth.MockAuthentication;
+
+import org.testng.annotations.Test;
+import org.testng.Assert;
+
+public class AuthenticationTest {
+
+    @Test
+    public void testConfigureDefaultFormat() {
+        try {
+            MockAuthentication testAuthentication = (MockAuthentication) AuthenticationFactory.create("org.apache.pulsar.client.impl.auth.MockAuthentication", "key1:value1,key2:value2");
+            Assert.assertEquals(testAuthentication.authParamsMap.get("key1"), "value1");
+            Assert.assertEquals(testAuthentication.authParamsMap.get("key2"), "value2");
+        } catch (PulsarClientException.UnsupportedAuthenticationException e) {
+            e.printStackTrace();
+            Assert.fail();
+        }
+    }
+
+    @Test
+    public void testConfigureWrongFormat() {
+        try {
+            MockAuthentication testAuthentication = (MockAuthentication) AuthenticationFactory.create("org.apache.pulsar.client.impl.auth.MockAuthentication", "foobar");
+            Assert.assertTrue(testAuthentication.authParamsMap.isEmpty());
+        } catch (PulsarClientException.UnsupportedAuthenticationException e) {
+            e.printStackTrace();
+            Assert.fail();
+        }
+    }
+
+    @Test
+    public void testConfigureNull() {
+        try {
+            MockAuthentication testAuthentication = (MockAuthentication) AuthenticationFactory.create("org.apache.pulsar.client.impl.auth.MockAuthentication", (String) null);
+            Assert.assertTrue(testAuthentication.authParamsMap.isEmpty());
+        } catch (PulsarClientException.UnsupportedAuthenticationException e) {
+            e.printStackTrace();
+            Assert.fail();
+        }
+    }
+
+    @Test
+    public void testConfigureEmpty() {
+        try {
+            MockAuthentication testAuthentication = (MockAuthentication) AuthenticationFactory.create("org.apache.pulsar.client.impl.auth.MockAuthentication", "");
+            Assert.assertTrue(testAuthentication.authParamsMap.isEmpty());
+        } catch (PulsarClientException.UnsupportedAuthenticationException e) {
+            e.printStackTrace();
+            Assert.fail();
+        }
+    }
+
+    @Test
+    public void testConfigurePluginSide() {
+        try {
+            MockEncodedAuthenticationParameterSupport testAuthentication = (MockEncodedAuthenticationParameterSupport) AuthenticationFactory.create("org.apache.pulsar.client.impl.auth.MockEncodedAuthenticationParameterSupport", "key1:value1;key2:value2");
+            Assert.assertEquals(testAuthentication.authParamsMap.get("key1"), "value1");
+            Assert.assertEquals(testAuthentication.authParamsMap.get("key2"), "value2");
+        } catch (PulsarClientException.UnsupportedAuthenticationException e) {
+            e.printStackTrace();
+            Assert.fail();
+        }
+    }
+}

pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/MockAuthentication.java

@@ -0,0 +1,56 @@
+/**
+ * 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.auth;
+
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.AuthenticationDataProvider;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+public class MockAuthentication implements Authentication {
+    public Map<String, String> authParamsMap = new HashMap<>();
+
+    @Override
+    public String getAuthMethodName() {
+        return null;
+    }
+
+    @Override
+    public AuthenticationDataProvider getAuthData() throws PulsarClientException {
+        return null;
+    }
+
+    @Override
+    public void configure(Map<String, String> authParams) {
+        authParamsMap = authParams;
+    }
+
+    @Override
+    public void start() throws PulsarClientException {
+
+    }
+
+    @Override
+    public void close() throws IOException {
+
+    }
+}

pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/MockEncodedAuthenticationParameterSupport.java

@@ -0,0 +1,68 @@
+/**
+ * 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.auth;
+
+import org.apache.pulsar.client.api.EncodedAuthenticationParameterSupport;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.AuthenticationDataProvider;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+public class MockEncodedAuthenticationParameterSupport implements Authentication, EncodedAuthenticationParameterSupport {
+    public Map<String, String> authParamsMap = new HashMap<>();
+
+    @Override
+    public void configure(String authParams) {
+        String[] params = authParams.split(";");
+        for (String p : params) {
+            String[] kv = p.split(":");
+            if (kv.length == 2) {
+                authParamsMap.put(kv[0], kv[1]);
+            }
+        }
+    }
+
+    @Override
+    public String getAuthMethodName() {
+        return null;
+    }
+
+    @Override
+    public AuthenticationDataProvider getAuthData() throws PulsarClientException {
+        return null;
+    }
+
+    @Override
+    public void configure(Map<String, String> authParams) {
+
+    }
+
+    @Override
+    public void start() throws PulsarClientException {
+
+    }
+
+    @Override
+    public void close() throws IOException {
+
+    }
+}