rdhabalia closed pull request #1935: Add Client auth plugin and tls support for
function to connect with broker
URL: https://github.com/apache/incubator-pulsar/pull/1935
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
index 5089e95b87..328d6b23c2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.io;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static
org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
import static org.mockito.Mockito.spy;
@@ -27,6 +28,8 @@
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URL;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -40,14 +43,17 @@
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.ServiceConfigurationUtils;
+import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl;
import org.apache.pulsar.client.admin.BrokerStats;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.auth.AuthenticationTls;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.SubscriptionStats;
import org.apache.pulsar.common.policies.data.TenantInfo;
@@ -87,7 +93,7 @@
ServiceConfiguration config;
WorkerConfig workerConfig;
- URL url;
+ URL urlTls;
PulsarService pulsar;
PulsarAdmin admin;
PulsarClient pulsarClient;
@@ -102,8 +108,16 @@
private final int ZOOKEEPER_PORT = PortManager.nextFreePort();
private final int brokerWebServicePort = PortManager.nextFreePort();
+ private final int brokerWebServiceTlsPort = PortManager.nextFreePort();
private final int brokerServicePort = PortManager.nextFreePort();
+ private final int brokerServiceTlsPort = PortManager.nextFreePort();
private final int workerServicePort = PortManager.nextFreePort();
+
+ private final String TLS_SERVER_CERT_FILE_PATH =
"./src/test/resources/authentication/tls/broker-cert.pem";
+ private final String TLS_SERVER_KEY_FILE_PATH =
"./src/test/resources/authentication/tls/broker-key.pem";
+ private final String TLS_CLIENT_CERT_FILE_PATH =
"./src/test/resources/authentication/tls/client-cert.pem";
+ private final String TLS_CLIENT_KEY_FILE_PATH =
"./src/test/resources/authentication/tls/client-key.pem";
+
private static final Logger log =
LoggerFactory.getLogger(PulsarSinkE2ETest.class);
@BeforeMethod
@@ -118,31 +132,67 @@ void setup(Method method) throws Exception {
bkEnsemble = new LocalBookkeeperEnsemble(3, ZOOKEEPER_PORT,
PortManager.nextFreePort());
bkEnsemble.start();
- String hostHttpUrl = "http://127.0.0.1"; + ":";
+ String brokerServiceUrl = "https://127.0.0.1:"; +
brokerWebServiceTlsPort;
config = spy(new ServiceConfiguration());
config.setClusterName("use");
+ Set<String> superUsers = Sets.newHashSet("superUser");
+ config.setSuperUserRoles(superUsers);
config.setWebServicePort(brokerWebServicePort);
+ config.setWebServicePortTls(brokerWebServiceTlsPort);
config.setZookeeperServers("127.0.0.1" + ":" + ZOOKEEPER_PORT);
config.setBrokerServicePort(brokerServicePort);
+ config.setBrokerServicePortTls(brokerServiceTlsPort);
config.setLoadManagerClassName(SimpleLoadManagerImpl.class.getName());
+
+
+ Set<String> providers = new HashSet<>();
+ providers.add(AuthenticationProviderTls.class.getName());
+ config.setAuthenticationEnabled(true);
+ config.setAuthenticationProviders(providers);
+ config.setTlsEnabled(true);
+ config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
+ config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
+ config.setTlsAllowInsecureConnection(true);
+
+
functionsWorkerService = createPulsarFunctionWorker(config);
- url = new URL(hostHttpUrl + brokerWebServicePort);
+ urlTls = new URL(brokerServiceUrl);
boolean isFunctionWebServerRequired = method.getName()
.equals("testExternalReplicatorRedirectionToWorkerService");
Optional<WorkerService> functionWorkerService =
isFunctionWebServerRequired ? Optional.ofNullable(null)
: Optional.of(functionsWorkerService);
pulsar = new PulsarService(config, functionWorkerService);
pulsar.start();
- admin = new PulsarAdmin(url, (Authentication) null);
+
+ Map<String, String> authParams = new HashMap<>();
+ authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH);
+ authParams.put("tlsKeyFile", TLS_CLIENT_KEY_FILE_PATH);
+ Authentication authTls = new AuthenticationTls();
+ authTls.configure(authParams);
+
+ admin = spy(
+
PulsarAdmin.builder().serviceHttpUrl(brokerServiceUrl).tlsTrustCertsFilePath(TLS_CLIENT_CERT_FILE_PATH)
+
.allowTlsInsecureConnection(true).authentication(authTls).build());
+
brokerStatsClient = admin.brokerStats();
primaryHost = String.format("http://%s:%d";,
InetAddress.getLocalHost().getHostName(), brokerWebServicePort);
// update cluster metadata
- ClusterData clusterData = new ClusterData(url.toString());
+ ClusterData clusterData = new ClusterData(urlTls.toString());
admin.clusters().updateCluster(config.getClusterName(), clusterData);
- pulsarClient =
PulsarClient.builder().serviceUrl(url.toString()).statsInterval(0,
TimeUnit.SECONDS).build();
+ ClientBuilder clientBuilder =
PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl());
+ if (isNotBlank(workerConfig.getClientAuthenticationPlugin())
+ &&
isNotBlank(workerConfig.getClientAuthenticationParameters())) {
+ clientBuilder.enableTls(workerConfig.isUseTls());
+
clientBuilder.allowTlsInsecureConnection(workerConfig.isTlsAllowInsecureConnection());
+
clientBuilder.authentication(workerConfig.getClientAuthenticationPlugin(),
+ workerConfig.getClientAuthenticationParameters());
+ }
+ pulsarClient = clientBuilder.build();
+ // pulsarClient =
PulsarClient.builder().serviceUrl(urlTls.toString()).statsInterval(0,
+ // TimeUnit.SECONDS).build();
TenantInfo propAdmin = new TenantInfo();
propAdmin.setAllowedClusters(Sets.newHashSet(Lists.newArrayList("use")));
@@ -166,6 +216,7 @@ void shutdown() throws Exception {
if (workerExecutor != null) {
workerExecutor.shutdown();
}
+ pulsarClient.close();
admin.close();
pulsar.close();
functionsWorkerService.stop();
@@ -179,8 +230,8 @@ private WorkerService
createPulsarFunctionWorker(ServiceConfiguration config) {
org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler.class.getName());
workerConfig.setThreadContainerFactory(new
WorkerConfig.ThreadContainerFactory().setThreadGroupName("use"));
// worker talks to local broker
- workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" +
config.getBrokerServicePort());
- workerConfig.setPulsarWebServiceUrl("http://127.0.0.1:"; +
config.getWebServicePort());
+ workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" +
config.getBrokerServicePortTls());
+ workerConfig.setPulsarWebServiceUrl("https://127.0.0.1:"; +
config.getWebServicePortTls());
workerConfig.setFailureCheckFreqMs(100);
workerConfig.setNumFunctionPackageReplicas(1);
workerConfig.setClusterCoordinationTopicName("coordinate");
@@ -193,11 +244,19 @@ private WorkerService
createPulsarFunctionWorker(ServiceConfiguration config) {
workerConfig.setWorkerHostname(hostname);
workerConfig
.setWorkerId("c-" + config.getClusterName() + "-fw-" +
hostname + "-" + workerConfig.getWorkerPort());
+
+
workerConfig.setClientAuthenticationPlugin(AuthenticationTls.class.getName());
+ workerConfig.setClientAuthenticationParameters(
+ String.format("tlsCertFile:%s,tlsKeyFile:%s",
TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH));
+ workerConfig.setUseTls(true);
+ workerConfig.setTlsAllowInsecureConnection(true);
+ workerConfig.setTlsTrustCertsFilePath(TLS_CLIENT_CERT_FILE_PATH);
+
return new WorkerService(workerConfig);
}
/**
- * Validates pulsar sink e2e functionality on functions.
+ * Validates pulsar sink e2e functionality on functions.
*
* @throws Exception
*/
@@ -236,8 +295,8 @@ public void testE2EPulsarSink() throws Exception {
}
retryStrategically((test) -> {
try {
- SubscriptionStats subStats =
admin.topics().getStats(sourceTopic).subscriptions.values()
- .iterator().next();
+ SubscriptionStats subStats =
admin.topics().getStats(sourceTopic).subscriptions.values().iterator()
+ .next();
return subStats.unackedMessages == 0;
} catch (PulsarAdminException e) {
return false;
@@ -245,8 +304,8 @@ public void testE2EPulsarSink() throws Exception {
}, 5, 150);
// validate pulsar-sink consumer has consumed all messages and
delivered to Pulsar sink but unacked messages
// due to publish failure
-
Assert.assertEquals(admin.topics().getStats(sourceTopic).subscriptions.values().iterator()
- .next().unackedMessages, 0);
+ Assert.assertEquals(
+
admin.topics().getStats(sourceTopic).subscriptions.values().iterator().next().unackedMessages,
0);
}
@@ -281,7 +340,7 @@ protected FunctionDetails createSinkConfig(String jarFile,
String tenant, String
// set up sink spec
SinkSpec.Builder sinkSpecBuilder = SinkSpec.newBuilder();
- //sinkSpecBuilder.setClassName(PulsarSink.class.getName());
+ // sinkSpecBuilder.setClassName(PulsarSink.class.getName());
sinkSpecBuilder.setTopic(String.format("persistent://%s/%s/%s",
tenant, namespace, "output"));
Map<String, Object> sinkConfigMap = Maps.newHashMap();
sinkSpecBuilder.setConfigs(new Gson().toJson(sinkConfigMap));
@@ -290,4 +349,4 @@ protected FunctionDetails createSinkConfig(String jarFile,
String tenant, String
return functionDetailsBuilder.build();
}
-}
+}
\ No newline at end of file
diff --git
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index 69157746ce..4b17ff5527 100644
---
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -51,6 +51,8 @@
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.internal.FunctionsImpl;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.proto.Function.Resources;
@@ -642,11 +644,34 @@ protected FunctionDetails convert(FunctionConfig
functionConfig)
@Parameter(names = "--brokerServiceUrl", description = "The URL for
the Pulsar broker")
protected String brokerServiceUrl;
+
+ @Parameter(names = "--clientAuthPlugin", description = "Client
authentication plugin using which function-process can connect to broker")
+ protected String clientAuthPlugin;
+
+ @Parameter(names = "--clientAuthParams", description = "Client
authentication param")
+ protected String clientAuthParams;
+
+ @Parameter(names = "--use_tls", description = "Use tls connection\n")
+ protected boolean useTls;
+
+ @Parameter(names = "--tls_allow_insecure", description = "Allow
insecure tls connection\n")
+ protected boolean tlsAllowInsecureConnection;
+
+ @Parameter(names = "--hostname_verification_enabled", description =
"Enable hostname verification")
+ protected boolean tlsHostNameVerificationEnabled;
+
+ @Parameter(names = "--tls_trust_cert_path", description = "tls trust
cert file path")
+ protected String tlsTrustCertFilePath;
@Override
void runCmd() throws Exception {
- CmdFunctions.startLocalRun(convertProto2(functionConfig),
- functionConfig.getParallelism(), brokerServiceUrl,
userCodeFile, admin);
+ CmdFunctions.startLocalRun(convertProto2(functionConfig),
functionConfig.getParallelism(), brokerServiceUrl,
+
AuthenticationConfig.builder().clientAuthenticationPlugin(clientAuthPlugin)
+
.clientAuthenticationParameters(clientAuthParams).useTls(useTls)
+
.tlsAllowInsecureConnection(tlsAllowInsecureConnection)
+
.tlsHostnameVerificationEnable(tlsHostNameVerificationEnabled)
+
.tlsTrustCertsFilePath(tlsTrustCertFilePath).build(),
+ userCodeFile, admin);
}
}
@@ -911,7 +936,8 @@ private void parseFullyQualifiedFunctionName(String fqfn,
FunctionConfig functio
}
protected static void
startLocalRun(org.apache.pulsar.functions.proto.Function.FunctionDetails
functionDetails,
- int parallelism, String
brokerServiceUrl, String userCodeFile, PulsarAdmin admin)
+ int parallelism, String brokerServiceUrl, AuthenticationConfig
authConfig,
+ String userCodeFile, PulsarAdmin admin)
throws Exception {
String serviceUrl = admin.getServiceUrl();
@@ -921,8 +947,8 @@ protected static void
startLocalRun(org.apache.pulsar.functions.proto.Function.F
if (serviceUrl == null) {
serviceUrl = DEFAULT_SERVICE_URL;
}
- try (ProcessRuntimeFactory containerFactory = new
ProcessRuntimeFactory(
- serviceUrl, null, null, null)) {
+ try (ProcessRuntimeFactory containerFactory = new
ProcessRuntimeFactory(serviceUrl, authConfig, null, null,
+ null)) {
List<RuntimeSpawner> spawners = new LinkedList<>();
for (int i = 0; i < parallelism; ++i) {
InstanceConfig instanceConfig = new InstanceConfig();
diff --git
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
index 6f67ee753a..59b9201ae4 100644
---
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
+++
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
@@ -29,6 +29,7 @@
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.internal.FunctionsImpl;
import org.apache.pulsar.functions.api.utils.IdentityFunction;
+import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.proto.Function.Resources;
@@ -99,11 +100,35 @@ void processArguments() throws Exception {
@Parameter(names = "--brokerServiceUrl", description = "The URL for
the Pulsar broker")
protected String brokerServiceUrl;
+
+ @Parameter(names = "--clientAuthPlugin", description = "Client
authentication plugin using which function-process can connect to broker")
+ protected String clientAuthPlugin;
+
+ @Parameter(names = "--clientAuthParams", description = "Client
authentication param")
+ protected String clientAuthParams;
+
+ @Parameter(names = "--use_tls", description = "Use tls connection\n")
+ protected boolean useTls;
+
+ @Parameter(names = "--tls_allow_insecure", description = "Allow
insecure tls connection\n")
+ protected boolean tlsAllowInsecureConnection;
+
+ @Parameter(names = "--hostname_verification_enabled", description =
"Enable hostname verification")
+ protected boolean tlsHostNameVerificationEnabled;
+
+ @Parameter(names = "--tls_trust_cert_path", description = "tls trust
cert file path")
+ protected String tlsTrustCertFilePath;
@Override
void runCmd() throws Exception {
- CmdFunctions.startLocalRun(createSinkConfigProto2(sinkConfig),
- sinkConfig.getParallelism(), brokerServiceUrl, jarFile,
admin);
+ CmdFunctions.startLocalRun(createSinkConfigProto2(sinkConfig),
sinkConfig.getParallelism(),
+ brokerServiceUrl,
+
AuthenticationConfig.builder().clientAuthenticationPlugin(clientAuthPlugin)
+
.clientAuthenticationParameters(clientAuthParams).useTls(useTls)
+
.tlsAllowInsecureConnection(tlsAllowInsecureConnection)
+
.tlsHostnameVerificationEnable(tlsHostNameVerificationEnabled)
+
.tlsTrustCertsFilePath(tlsTrustCertFilePath).build(),
+ jarFile, admin);
}
}
diff --git
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
index 8348382fbf..a3bec4aa7a 100644
---
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
+++
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
@@ -29,6 +29,7 @@
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.internal.FunctionsImpl;
import org.apache.pulsar.functions.api.utils.IdentityFunction;
+import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.proto.Function.Resources;
import org.apache.pulsar.functions.proto.Function.SinkSpec;
@@ -95,11 +96,35 @@ void processArguments() throws Exception {
@Parameter(names = "--brokerServiceUrl", description = "The URL for
the Pulsar broker")
protected String brokerServiceUrl;
+
+ @Parameter(names = "--clientAuthPlugin", description = "Client
authentication plugin using which function-process can connect to broker")
+ protected String clientAuthPlugin;
+
+ @Parameter(names = "--clientAuthParams", description = "Client
authentication param")
+ protected String clientAuthParams;
+
+ @Parameter(names = "--use_tls", description = "Use tls connection\n")
+ protected boolean useTls;
+
+ @Parameter(names = "--tls_allow_insecure", description = "Allow
insecure tls connection\n")
+ protected boolean tlsAllowInsecureConnection;
+
+ @Parameter(names = "--hostname_verification_enabled", description =
"Enable hostname verification")
+ protected boolean tlsHostNameVerificationEnabled;
+
+ @Parameter(names = "--tls_trust_cert_path", description = "tls trust
cert file path")
+ protected String tlsTrustCertFilePath;
@Override
void runCmd() throws Exception {
- CmdFunctions.startLocalRun(createSourceConfigProto2(sourceConfig),
- sourceConfig.getParallelism(), brokerServiceUrl, jarFile,
admin);
+ CmdFunctions.startLocalRun(createSourceConfigProto2(sourceConfig),
sourceConfig.getParallelism(),
+ brokerServiceUrl,
+
AuthenticationConfig.builder().clientAuthenticationPlugin(clientAuthPlugin)
+
.clientAuthenticationParameters(clientAuthParams).useTls(useTls)
+
.tlsAllowInsecureConnection(tlsAllowInsecureConnection)
+
.tlsHostnameVerificationEnable(tlsHostNameVerificationEnabled)
+
.tlsTrustCertsFilePath(tlsTrustCertFilePath).build(),
+ jarFile, admin);
}
}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
index 1d3280b7ba..cedc892cab 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
@@ -55,7 +55,7 @@
private int maxLookupRequest = 50000;
private int maxNumberOfRejectedRequestPerConnection = 50;
private int keepAliveIntervalSeconds = 30;
-
+
public ClientConfigurationData clone() {
try {
return (ClientConfigurationData) super.clone();
diff --git
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/AuthenticationConfig.java
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/AuthenticationConfig.java
new file mode 100644
index 0000000000..f4c78eac66
--- /dev/null
+++
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/AuthenticationConfig.java
@@ -0,0 +1,44 @@
+/**
+ * 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.functions.instance;
+
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+
+/**
+ * Configuration to aggregate various authentication params.
+ */
+@Data
+@Getter
+@Setter
+@EqualsAndHashCode
+@ToString
+@Builder
+public class AuthenticationConfig {
+ private String clientAuthenticationPlugin;
+ private String clientAuthenticationParameters;
+ private String tlsTrustCertsFilePath;
+ private boolean useTls;
+ private boolean tlsAllowInsecureConnection;
+ private boolean tlsHostnameVerificationEnable;
+}
diff --git a/pulsar-functions/instance/src/main/python/python_instance_main.py
b/pulsar-functions/instance/src/main/python/python_instance_main.py
index b3eaa1c7ba..a3c1179b55 100644
--- a/pulsar-functions/instance/src/main/python/python_instance_main.py
+++ b/pulsar-functions/instance/src/main/python/python_instance_main.py
@@ -30,6 +30,7 @@
import time
import json
+from pulsar import Authentication
import pulsar
import Function_pb2
@@ -63,6 +64,12 @@ def main():
parser.add_argument('--function_version', required=True, help='Function
Version')
parser.add_argument('--processing_guarantees', required=True,
help='Processing Guarantees')
parser.add_argument('--pulsar_serviceurl', required=True, help='Pulsar
Service Url')
+ parser.add_argument('--client_auth_plugin', required=False, help='Client
authentication plugin')
+ parser.add_argument('--client_auth_params', required=False, help='Client
authentication params')
+ parser.add_argument('--use_tls', required=False, help='Use tls')
+ parser.add_argument('--tls_allow_insecure_connection', required=False,
help='Tls allow insecure connection')
+ parser.add_argument('--hostname_verification_enabled', required=False,
help='Enable hostname verification')
+ parser.add_argument('--tls_trust_cert_path', required=False, help='Tls trust
cert file path')
parser.add_argument('--port', required=True, help='Instance Port', type=int)
parser.add_argument('--max_buffered_tuples', required=True, help='Maximum
number of Buffered tuples')
parser.add_argument('--user_config', required=False, help='User Config')
@@ -124,7 +131,19 @@ def main():
if args.user_config != None and len(args.user_config) != 0:
function_details.userConfig = args.user_config
- pulsar_client = pulsar.Client(args.pulsar_serviceurl)
+ authentication = None
+ use_tls = False
+ tls_allow_insecure_connection = False
+ tls_trust_cert_path = None
+ if args.client_auth_plugin and args.client_auth_params:
+ authentication = pulsar.Authentication(args.client_auth_plugin,
args.client_auth_params)
+ if args.use_tls == "true":
+ use_tls = True
+ if args.tls_allow_insecure_connection == "true":
+ tls_allow_insecure_connection = True
+ if args.tls_trust_cert_path:
+ tls_trust_cert_path = args.tls_trust_cert_path
+ pulsar_client = pulsar.Client(args.pulsar_serviceurl, authentication, 30, 1,
1, 50000, None, use_tls, tls_trust_cert_path, tls_allow_insecure_connection)
pyinstance = python_instance.PythonInstance(str(args.instance_id),
str(args.function_id),
str(args.function_version),
function_details,
int(args.max_buffered_tuples),
str(args.py),
diff --git
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
index b602a1d263..554f9acb99 100644
---
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
+++
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
@@ -30,6 +30,8 @@
import lombok.extern.slf4j.Slf4j;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+
+import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.ProcessingGuarantees;
@@ -39,7 +41,6 @@
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceControlGrpc;
-import java.lang.reflect.Type;
import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.ExecutionException;
@@ -82,7 +83,25 @@
@Parameter(names = "--pulsar_serviceurl", description = "Pulsar Service
Url\n", required = true)
protected String pulsarServiceUrl;
-
+
+ @Parameter(names = "--client_auth_plugin", description = "Client auth
plugin name\n")
+ protected String clientAuthenticationPlugin;
+
+ @Parameter(names = "--client_auth_params", description = "Client auth
param\n")
+ protected String clientAuthenticationParameters;
+
+ @Parameter(names = "--use_tls", description = "Use tls connection\n")
+ protected String useTls = Boolean.FALSE.toString();
+
+ @Parameter(names = "--tls_allow_insecure", description = "Allow insecure
tls connection\n")
+ protected String tlsAllowInsecureConnection = Boolean.TRUE.toString();
+
+ @Parameter(names = "--hostname_verification_enabled", description =
"Enable hostname verification")
+ protected String tlsHostNameVerificationEnabled = Boolean.FALSE.toString();
+
+ @Parameter(names = "--tls_trust_cert_path", description = "tls trust cert
file path")
+ protected String tlsTrustCertFilePath;
+
@Parameter(names = "--state_storage_serviceurl", description = "State
Storage Service Url\n", required= false)
protected String stateStorageServiceUrl;
@@ -96,7 +115,7 @@
protected String userConfig;
@Parameter(names = "--auto_ack", description = "Enable Auto Acking?\n")
- protected String autoAck = "true";
+ protected String autoAck = Boolean.TRUE.toString();
@Parameter(names = "--source_classname", description = "The source
classname")
protected String sourceClassname;
@@ -158,11 +177,7 @@ public void start() throws Exception {
functionDetailsBuilder.setLogTopic(logTopic);
}
functionDetailsBuilder.setProcessingGuarantees(processingGuarantees);
- if (autoAck.equals("true")) {
- functionDetailsBuilder.setAutoAck(true);
- } else {
- functionDetailsBuilder.setAutoAck(false);
- }
+ functionDetailsBuilder.setAutoAck(isTrue(autoAck));
if (userConfig != null && !userConfig.isEmpty()) {
functionDetailsBuilder.setUserConfig(userConfig);
}
@@ -207,10 +222,13 @@ public void start() throws Exception {
instanceConfig.setFunctionDetails(functionDetails);
instanceConfig.setPort(port);
- ThreadRuntimeFactory containerFactory = new ThreadRuntimeFactory(
- "LocalRunnerThreadGroup",
- pulsarServiceUrl,
- stateStorageServiceUrl);
+ ThreadRuntimeFactory containerFactory = new
ThreadRuntimeFactory("LocalRunnerThreadGroup", pulsarServiceUrl,
+ stateStorageServiceUrl,
+
AuthenticationConfig.builder().clientAuthenticationPlugin(clientAuthenticationPlugin)
+
.clientAuthenticationParameters(clientAuthenticationParameters).useTls(isTrue(useTls))
+
.tlsAllowInsecureConnection(isTrue(tlsAllowInsecureConnection))
+
.tlsHostnameVerificationEnable(isTrue(tlsHostNameVerificationEnabled))
+ .tlsTrustCertsFilePath(tlsTrustCertFilePath).build());
runtimeSpawner = new RuntimeSpawner(
instanceConfig,
jarFile,
@@ -257,6 +275,10 @@ public void run() {
close();
}
+ private static boolean isTrue(String param) {
+ return Boolean.TRUE.toString().equals(param);
+ }
+
public static void main(String[] args) throws Exception {
JavaInstanceMain javaInstanceMain = new JavaInstanceMain();
JCommander jcommander = new JCommander(javaInstanceMain);
diff --git
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java
index 715eab62a7..e94e39ddae 100644
---
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java
+++
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java
@@ -30,6 +30,9 @@
import lombok.extern.slf4j.Slf4j;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.InstanceCommunication;
@@ -66,17 +69,20 @@
String instanceFile,
String logDirectory,
String codeFile,
- String pulsarServiceUrl) {
+ String pulsarServiceUrl,
+ AuthenticationConfig authConfig) {
this.instanceConfig = instanceConfig;
this.instancePort = instanceConfig.getPort();
- this.processArgs = composeArgs(instanceConfig, instanceFile,
logDirectory, codeFile, pulsarServiceUrl);
+ this.processArgs = composeArgs(instanceConfig, instanceFile,
logDirectory, codeFile, pulsarServiceUrl,
+ authConfig);
}
private List<String> composeArgs(InstanceConfig instanceConfig,
String instanceFile,
String logDirectory,
String codeFile,
- String pulsarServiceUrl) {
+ String pulsarServiceUrl,
+ AuthenticationConfig authConfig) {
List<String> args = new LinkedList<>();
if (instanceConfig.getFunctionDetails().getRuntime() ==
Function.FunctionDetails.Runtime.JAVA) {
args.add("java");
@@ -135,6 +141,25 @@
args.add(String.valueOf(instanceConfig.getFunctionDetails().getProcessingGuarantees()));
args.add("--pulsar_serviceurl");
args.add(pulsarServiceUrl);
+ if (authConfig != null) {
+ if (isNotBlank(authConfig.getClientAuthenticationPlugin())
+ &&
isNotBlank(authConfig.getClientAuthenticationParameters())) {
+ args.add("--client_auth_plugin");
+ args.add(authConfig.getClientAuthenticationPlugin());
+ args.add("--client_auth_params");
+ args.add(authConfig.getClientAuthenticationParameters());
+ }
+ args.add("--use_tls");
+ args.add(Boolean.toString(authConfig.isUseTls()));
+ args.add("--tls_allow_insecure");
+
args.add(Boolean.toString(authConfig.isTlsAllowInsecureConnection()));
+ args.add("--hostname_verification_enabled");
+
args.add(Boolean.toString(authConfig.isTlsHostnameVerificationEnable()));
+ if(isNotBlank(authConfig.getTlsTrustCertsFilePath())) {
+ args.add("--tls_trust_cert_path");
+ args.add(authConfig.getTlsTrustCertsFilePath());
+ }
+ }
args.add("--max_buffered_tuples");
args.add(String.valueOf(instanceConfig.getMaxBufferedTuples()));
String userConfig =
instanceConfig.getFunctionDetails().getUserConfig();
diff --git
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntimeFactory.java
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntimeFactory.java
index 2223cb7cad..8fc5b90033 100644
---
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntimeFactory.java
+++
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntimeFactory.java
@@ -21,6 +21,8 @@
import com.google.common.annotations.VisibleForTesting;
import lombok.extern.slf4j.Slf4j;
+
+import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import java.nio.file.Paths;
@@ -32,17 +34,20 @@
public class ProcessRuntimeFactory implements RuntimeFactory {
private String pulsarServiceUrl;
+ private AuthenticationConfig authConfig;
private String javaInstanceJarFile;
private String pythonInstanceFile;
private String logDirectory;
@VisibleForTesting
public ProcessRuntimeFactory(String pulsarServiceUrl,
+ AuthenticationConfig authConfig,
String javaInstanceJarFile,
String pythonInstanceFile,
String logDirectory) {
this.pulsarServiceUrl = pulsarServiceUrl;
+ this.authConfig = authConfig;
this.javaInstanceJarFile = javaInstanceJarFile;
this.pythonInstanceFile = pythonInstanceFile;
this.logDirectory = logDirectory;
@@ -100,7 +105,8 @@ public ProcessRuntime createContainer(InstanceConfig
instanceConfig, String code
instanceFile,
logDirectory,
codeFile,
- pulsarServiceUrl);
+ pulsarServiceUrl,
+ authConfig);
}
@Override
diff --git
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntimeFactory.java
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntimeFactory.java
index df89eed8ea..f9c9cff3f8 100644
---
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntimeFactory.java
+++
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntimeFactory.java
@@ -23,8 +23,13 @@
import lombok.extern.slf4j.Slf4j;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+
+import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
+import
org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
+import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.utils.functioncache.FunctionCacheManager;
import
org.apache.pulsar.functions.utils.functioncache.FunctionCacheManagerImpl;
@@ -41,26 +46,40 @@
private final String storageServiceUrl;
private volatile boolean closed;
- public ThreadRuntimeFactory(String threadGroupName,
- String pulsarServiceUrl,
- String storageServiceUrl)
- throws Exception {
- this(
- threadGroupName,
- pulsarServiceUrl != null ?
PulsarClient.builder().serviceUrl(pulsarServiceUrl).build() : null,
- storageServiceUrl);
+ public ThreadRuntimeFactory(String threadGroupName, String
pulsarServiceUrl, String storageServiceUrl,
+ AuthenticationConfig authConfig) throws Exception {
+ this(threadGroupName, createPulsarClient(pulsarServiceUrl,
authConfig), storageServiceUrl);
}
@VisibleForTesting
- ThreadRuntimeFactory(String threadGroupName,
- PulsarClient pulsarClient,
- String storageServiceUrl) {
+ ThreadRuntimeFactory(String threadGroupName, PulsarClient pulsarClient,
String storageServiceUrl) {
this.fnCache = new FunctionCacheManagerImpl();
this.threadGroup = new ThreadGroup(threadGroupName);
this.pulsarClient = pulsarClient;
this.storageServiceUrl = storageServiceUrl;
}
+ private static PulsarClient createPulsarClient(String pulsarServiceUrl,
AuthenticationConfig authConfig)
+ throws PulsarClientException {
+ ClientBuilder clientBuilder = null;
+ if (isNotBlank(pulsarServiceUrl)) {
+ clientBuilder =
PulsarClient.builder().serviceUrl(pulsarServiceUrl);
+ if (authConfig != null) {
+ if (isNotBlank(authConfig.getClientAuthenticationPlugin())
+ &&
isNotBlank(authConfig.getClientAuthenticationParameters())) {
+
clientBuilder.authentication(authConfig.getClientAuthenticationPlugin(),
+ authConfig.getClientAuthenticationParameters());
+ }
+ clientBuilder.enableTls(authConfig.isUseTls());
+
clientBuilder.allowTlsInsecureConnection(authConfig.isTlsAllowInsecureConnection());
+
clientBuilder.enableTlsHostnameVerification(authConfig.isTlsHostnameVerificationEnable());
+
clientBuilder.tlsTrustCertsFilePath(authConfig.getTlsTrustCertsFilePath());
+ }
+ return clientBuilder.build();
+ }
+ return null;
+ }
+
@Override
public ThreadRuntime createContainer(InstanceConfig instanceConfig, String
jarFile) {
return new ThreadRuntime(
diff --git
a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java
b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java
index 65e8d89368..675f3de0d3 100644
---
a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java
+++
b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java
@@ -66,7 +66,7 @@ public ProcessRuntimeTest() {
this.pulsarServiceUrl = "pulsar://localhost:6670";
this.logDirectory = "Users/user/logs";
this.factory = new ProcessRuntimeFactory(
- pulsarServiceUrl, javaInstanceJarFile, pythonInstanceFile,
logDirectory);
+ pulsarServiceUrl, null, javaInstanceJarFile, pythonInstanceFile,
logDirectory);
}
@AfterMethod
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
index af23992308..ba1a33d997 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
@@ -26,6 +26,7 @@
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.functions.proto.Function.Assignment;
+import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.Request.AssignmentsUpdate;
import org.apache.pulsar.functions.runtime.RuntimeFactory;
@@ -92,14 +93,23 @@ public FunctionRuntimeManager(WorkerConfig workerConfig,
this.functionAssignmentTailer = new FunctionAssignmentTailer(this,
reader);
+ AuthenticationConfig authConfig = AuthenticationConfig.builder()
+
.clientAuthenticationPlugin(workerConfig.getClientAuthenticationPlugin())
+
.clientAuthenticationParameters(workerConfig.getClientAuthenticationParameters())
+ .tlsTrustCertsFilePath(workerConfig.getTlsTrustCertsFilePath())
+
.useTls(workerConfig.isUseTls()).tlsAllowInsecureConnection(workerConfig.isTlsAllowInsecureConnection())
+
.tlsHostnameVerificationEnable(workerConfig.isTlsHostnameVerificationEnable()).build();
+
if (workerConfig.getThreadContainerFactory() != null) {
this.runtimeFactory = new ThreadRuntimeFactory(
workerConfig.getThreadContainerFactory().getThreadGroupName(),
workerConfig.getPulsarServiceUrl(),
- workerConfig.getStateStorageServiceUrl());
+ workerConfig.getStateStorageServiceUrl(),
+ authConfig);
} else if (workerConfig.getProcessContainerFactory() != null) {
this.runtimeFactory = new ProcessRuntimeFactory(
workerConfig.getPulsarServiceUrl(),
+ authConfig,
workerConfig.getProcessContainerFactory().getJavaInstanceJarLocation(),
workerConfig.getProcessContainerFactory().getPythonInstanceLocation(),
workerConfig.getProcessContainerFactory().getLogDirectory());
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/MembershipManager.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/MembershipManager.java
index dde017fe96..de47a33f56 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/MembershipManager.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/MembershipManager.java
@@ -280,7 +280,9 @@ public void checkFailures(FunctionMetaDataManager
functionMetaDataManager,
private PulsarAdmin getPulsarAdminClient() {
if (this.pulsarAdminClient == null) {
- this.pulsarAdminClient =
Utils.getPulsarAdminClient(this.workerConfig.getPulsarWebServiceUrl());
+ this.pulsarAdminClient =
Utils.getPulsarAdminClient(this.workerConfig.getPulsarWebServiceUrl(),
+ workerConfig.getClientAuthenticationPlugin(),
workerConfig.getClientAuthenticationParameters(),
+ workerConfig.getTlsTrustCertsFilePath(),
workerConfig.isTlsAllowInsecureConnection());
}
return this.pulsarAdminClient;
}
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
index 51de807abe..576257b3d6 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
@@ -35,7 +35,7 @@
import java.util.UUID;
import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
import org.apache.distributedlog.AppendOnlyStreamWriter;
import org.apache.distributedlog.DistributedLogConfiguration;
import org.apache.distributedlog.api.DistributedLogManager;
@@ -44,6 +44,7 @@
import org.apache.distributedlog.impl.metadata.BKDLConfig;
import org.apache.distributedlog.metadata.DLMetadata;
import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.functions.worker.dlog.DLInputStream;
import org.apache.pulsar.functions.worker.dlog.DLOutputStream;
@@ -212,9 +213,17 @@ public static URI initializeDlogNamespace(String
zkServers, String ledgersRootPa
return dlogUri;
}
- public static PulsarAdmin getPulsarAdminClient(String pulsarWebServiceUrl)
{
+ public static PulsarAdmin getPulsarAdminClient(String pulsarWebServiceUrl,
String authPlugin, String authParams, String tlsTrustCertsFilePath, boolean
allowTlsInsecureConnection) {
try {
- return
PulsarAdmin.builder().serviceHttpUrl(pulsarWebServiceUrl).build();
+ PulsarAdminBuilder adminBuilder =
PulsarAdmin.builder().serviceHttpUrl(pulsarWebServiceUrl);
+ if (isNotBlank(authPlugin) && isNotBlank(authParams)) {
+ adminBuilder.authentication(authPlugin, authParams);
+ }
+ if (isNotBlank(tlsTrustCertsFilePath)) {
+ adminBuilder.tlsTrustCertsFilePath(tlsTrustCertsFilePath);
+ }
+
adminBuilder.allowTlsInsecureConnection(allowTlsInsecureConnection);
+ return adminBuilder.build();
} catch (PulsarClientException e) {
log.error("Error creating pulsar admin client", e);
throw new RuntimeException(e);
@@ -235,7 +244,7 @@ public static String getFullyQualifiedInstanceId(String
tenant, String namespace
}
public static boolean isFunctionPackageUrlSupported(String functionPkgUrl)
{
- return StringUtils.isNotBlank(functionPkgUrl)
+ return isNotBlank(functionPkgUrl)
&& (functionPkgUrl.startsWith(Utils.HTTP) ||
functionPkgUrl.startsWith(Utils.FILE));
}
}
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java
index 4f700ab451..6e87b94cb4 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java
@@ -71,7 +71,9 @@ protected void doStartImpl() throws InterruptedException,
IOException, PulsarAdm
private static URI initialize(WorkerConfig workerConfig)
throws InterruptedException, PulsarAdminException, IOException {
// initializing pulsar functions namespace
- PulsarAdmin admin =
Utils.getPulsarAdminClient(workerConfig.getPulsarWebServiceUrl());
+ PulsarAdmin admin =
Utils.getPulsarAdminClient(workerConfig.getPulsarWebServiceUrl(),
+ workerConfig.getClientAuthenticationPlugin(),
workerConfig.getClientAuthenticationParameters(),
+ workerConfig.getTlsTrustCertsFilePath(),
workerConfig.isTlsAllowInsecureConnection());
InternalConfigurationData internalConf;
// make sure pulsar broker is up
log.info("Checking if pulsar service at {} is up...",
workerConfig.getPulsarWebServiceUrl());
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
index 535a7dd332..efbd0fdd84 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
@@ -63,7 +63,13 @@
private int initialBrokerReconnectMaxRetries;
private int assignmentWriteMaxRetries;
private long instanceLivenessCheckFreqMs;
-
+ private String clientAuthenticationPlugin;
+ private String clientAuthenticationParameters;
+ private boolean useTls = false;
+ private String tlsTrustCertsFilePath = "";
+ private boolean tlsAllowInsecureConnection = false;
+ private boolean tlsHostnameVerificationEnable = false;
+
@Data
@Setter
@Getter
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java
index dfe5834e27..4740d69308 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java
@@ -23,9 +23,12 @@
import java.net.URI;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
import org.apache.distributedlog.DistributedLogConfiguration;
import org.apache.distributedlog.api.namespace.Namespace;
import org.apache.distributedlog.api.namespace.NamespaceBuilder;
+import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
@@ -77,7 +80,17 @@ public void start(URI dlogUri) throws InterruptedException {
// initialize the function metadata manager
try {
- this.client =
PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl()).build();
+ ClientBuilder clientBuilder =
PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl());
+ if (isNotBlank(workerConfig.getClientAuthenticationPlugin())
+ &&
isNotBlank(workerConfig.getClientAuthenticationParameters())) {
+
clientBuilder.authentication(workerConfig.getClientAuthenticationPlugin(),
+ workerConfig.getClientAuthenticationParameters());
+ }
+ clientBuilder.enableTls(workerConfig.isUseTls());
+
clientBuilder.allowTlsInsecureConnection(workerConfig.isTlsAllowInsecureConnection());
+
clientBuilder.tlsTrustCertsFilePath(workerConfig.getTlsTrustCertsFilePath());
+
clientBuilder.enableTlsHostnameVerification(workerConfig.isTlsHostnameVerificationEnable());
+ this.client = clientBuilder.build();
log.info("Created Pulsar client");
//create scheduler manager
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
