sql support pulsar.client configuration

Author: gavingaozhangmin

pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/internal/PulsarClientUtils.java

@@ -24,9 +24,13 @@
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
+import org.apache.pulsar.shade.com.google.common.collect.Maps;
 import org.apache.pulsar.shade.org.apache.commons.lang3.StringUtils;
 
+import java.util.Map;
 import java.util.Properties;
+import java.util.stream.Collectors;
 
 /** Utility to create Pulsar Admin Client from adminUrl and clientConfigurationData. */
 public class PulsarClientUtils {
@@ -56,13 +60,18 @@ private static Authentication getAuth(ClientConfigurationData conf)
     }
 
     public static ClientConfigurationData newClientConf(String serviceUrl, Properties properties) {
+        Map<String, Object> clientConfData = getClientParams(Maps.fromProperties(properties));
         ClientConfigurationData clientConf = new ClientConfigurationData();
+        clientConf = ConfigurationDataUtils.loadData(clientConfData, clientConf, ClientConfigurationData.class);
         clientConf.setServiceUrl(serviceUrl);
-        if (properties != null) {
-            clientConf.setAuthParams(properties.getProperty(PulsarOptions.AUTH_PARAMS_KEY));
-            clientConf.setAuthPluginClassName(
-                    properties.getProperty(PulsarOptions.AUTH_PLUGIN_CLASSNAME_KEY));
-        }
         return clientConf;
     }
+
+    public static Map<String, Object> getClientParams(Map<String, String> parameters) {
+        return parameters.keySet().stream()
+                .filter(k -> k.startsWith(PulsarOptions.PULSAR_CLIENT_OPTION_KEY_PREFIX))
+                .collect(Collectors.toMap(
+                        k -> k.substring(PulsarOptions.PULSAR_CLIENT_OPTION_KEY_PREFIX.length()),
+                        k -> parameters.get(k)));
+    }
 }