Before using Pulsar Functions, you need to learn how to set up Pulsar Functions worker and how to configure Functions runtime.
Pulsar functions-worker
is a logic component to run Pulsar Functions in cluster mode. Two options are available, and you can select either based on your requirements.
Note
The--- Service Urls---
lines in the following diagrams represent Pulsar service URLs that Pulsar client and admin use to connect to a Pulsar cluster.
The following diagram illustrates the deployment of functions-workers running along with brokers.
To enable functions-worker running as part of a broker, you need to set functionsWorkerEnabled
to true
in the broker.conf
file.
functionsWorkerEnabled=true
If the functionsWorkerEnabled
is set to true
, the functions-worker is started as part of a broker. You need to configure the conf/functions_worker.yml
file to customize your functions_worker.
Before you run Functions-worker with broker, you have to configure Functions-worker, and then start it with brokers.
In this mode, most of the settings are already inherited from your broker configuration (for example, configurationStore settings, authentication settings, and so on) since functions-worker
is running as part of the broker.
Pay attention to the following required settings when configuring functions-worker in this mode.
numFunctionPackageReplicas
: The number of replicas to store function packages. The default value is 1
, which is good for standalone deployment. For production deployment, to ensure high availability, set it to be larger than 2
.initializedDlogMetadata
: Whether to initialize distributed log metadata in runtime. If it is set to true
, you must ensure that it has been initialized by bin/pulsar initialize-cluster-metadata
command.If authentication is enabled on the BookKeeper cluster, configure the following BookKeeper authentication settings.
bookkeeperClientAuthenticationPlugin
: the BookKeeper client authentication plugin name.bookkeeperClientAuthenticationParametersName
: the BookKeeper client authentication plugin parameters name.bookkeeperClientAuthenticationParameters
: the BookKeeper client authentication plugin parameters.If you want to use Stateful-Functions related functions (for example, putState()
and queryState()
related interfaces), follow steps below.
Enable the streamStorage service in the BookKeeper.
Currently, the service uses the NAR package, so you need to set the configuration in bookkeeper.conf
.
extraServerComponents=org.apache.bookkeeper.stream.server.StreamStorageLifecycleComponent
After starting bookie, use the following methods to check whether the streamStorage service is started correctly.
Input:
telnet localhost 4181
Output:
Trying 127.0.0.1... Connected to localhost. Escape character is '^]'.
Turn on this function in functions_worker.yml
.
stateStorageServiceUrl: bk://<bk-service-url>:4181
bk-service-url
is the service URL pointing to the BookKeeper table service.
Once you have configured the functions_worker.yml
file, you can start or restart your broker.
And then you can use the following command to verify if functions-worker
is running well.
curl <broker-ip>:8080/admin/v2/worker/cluster
After entering the command above, a list of active function workers in the cluster is returned. The output is similar to the following.
[{"workerId":"<worker-id>","workerHostname":"<worker-hostname>","port":8080}]
This section illustrates how to run functions-worker
as a separate process in separate machines.
Note
In this mode, make surefunctionsWorkerEnabled
is set tofalse
, so you won't startfunctions-worker
with brokers by mistake. Also, while accessing thefunctions-worker
to manage any of the functions, thepulsar-admin
CLI tool or any of the clients should use theworkerHostname
andworkerPort
that you set in Worker parameters to generate an--admin-url
.
To run function-worker separately, you have to configure the following parameters.
workerId
: The type is string. It is unique across clusters, which is used to identify a worker machine.workerHostname
: The hostname of the worker machine.workerPort
: The port that the worker server listens on. Keep it as default if you don't customize it.workerPortTls
: The TLS port that the worker server listens on. Keep it as default if you don't customize it.numFunctionPackageReplicas
: The number of replicas to store function packages. The default value is 1
.pulsarServiceUrl
: The Pulsar service URL for your broker cluster.pulsarWebServiceUrl
: The Pulsar web service URL for your broker cluster.pulsarFunctionsCluster
: Set the value to your Pulsar cluster name (same as the clusterName
setting in the broker configuration).If authentication is enabled for your broker cluster, you should configure the authentication plugin and parameters for the functions worker to communicate with the brokers.
clientAuthenticationPlugin
clientAuthenticationParameters
If you want to enable security on functions workers, you should:
To enable TLS transport encryption, configure the following settings.
useTLS: true pulsarServiceUrl: pulsar+ssl://localhost:6651/ pulsarWebServiceUrl: https://localhost:8443 tlsEnabled: true tlsCertificateFilePath: /path/to/functions-worker.cert.pem tlsKeyFilePath: /path/to/functions-worker.key-pk8.pem tlsTrustCertsFilePath: /path/to/ca.cert.pem // The path to trusted certificates used by the Pulsar client to authenticate with Pulsar brokers brokerClientTrustCertsFilePath: /path/to/ca.cert.pem
For details on TLS encryption, refer to Transport Encryption using TLS.
To enable authentication on Functions Worker, you need to configure the following settings.
Note
Substitute the providers list with the providers you want to enable.
authenticationEnabled: true authenticationProviders: [ provider1, provider2 ]
For TLS Authentication provider, follow the example below to add the necessary settings. See TLS Authentication for more details.
brokerClientAuthenticationPlugin: org.apache.pulsar.client.impl.auth.AuthenticationTls brokerClientAuthenticationParameters: tlsCertFile:/path/to/admin.cert.pem,tlsKeyFile:/path/to/admin.key-pk8.pem authenticationEnabled: true authenticationProviders: ['org.apache.pulsar.broker.authentication.AuthenticationProviderTls']
For SASL Authentication provider, add saslJaasClientAllowedIds
and saslJaasServerSectionName
under properties
if needed.
properties: saslJaasClientAllowedIds: .*pulsar.* saslJaasServerSectionName: Broker
For Token Authentication provider, add necessary settings for properties
if needed. See Token Authentication for more details. Note: key files must be DER-encoded
properties: tokenSecretKey: file://my/secret.key # If using public/private # tokenPublicKey: file:///path/to/public.key
To enable authorization on Functions Worker, you need to configure authorizationEnabled
, authorizationProvider
and configurationStoreServers
. The authentication provider connects to configurationStoreServers
to receive namespace policies.
authorizationEnabled: true authorizationProvider: org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider configurationStoreServers: <configuration-store-servers>
You should also configure a list of superuser roles. The superuser roles are able to access any admin API. The following is a configuration example.
superUserRoles: - role1 - role2 - role3
You can use the public and private key pair that the application configures to perform encryption. Only the consumers with a valid key can decrypt the encrypted messages.
To enable End-to-End encryption on Functions Worker, you can set it by specifying --producer-config
in the command line terminal, for more information, please refer to here.
We include the relevant configuration information of CryptoConfig
into ProducerConfig
. The specific configurable field information about CryptoConfig
is as follows:
public class CryptoConfig { private String cryptoKeyReaderClassName; private Map<String, Object> cryptoKeyReaderConfig; private String[] encryptionKeys; private ProducerCryptoFailureAction producerCryptoFailureAction; private ConsumerCryptoFailureAction consumerCryptoFailureAction; }
producerCryptoFailureAction
: define the action if producer fail to encrypt data one of FAIL
, SEND
.consumerCryptoFailureAction
: define the action if consumer fail to decrypt data one of FAIL
, DISCARD
, CONSUME
.If authentication is enabled on the BookKeeper cluster, you need configure the BookKeeper authentication settings as follows:
bookkeeperClientAuthenticationPlugin
: the plugin name of BookKeeper client authentication.bookkeeperClientAuthenticationParametersName
: the plugin parameters name of BookKeeper client authentication.bookkeeperClientAuthenticationParameters
: the plugin parameters of BookKeeper client authentication.Once you have finished configuring the functions_worker.yml
configuration file, you can start a functions-worker
in the background by using nohup with the pulsar-daemon
CLI tool:
bin/pulsar-daemon start functions-worker
You can also start functions-worker
in the foreground by using pulsar
CLI tool:
bin/pulsar functions-worker
When you are running functions-worker
in a separate cluster, the admin rest endpoints are split into two clusters. functions
, function-worker
, source
and sink
endpoints are now served by the functions-worker
cluster, while all the other remaining endpoints are served by the broker cluster. Hence you need to configure your pulsar-admin
to use the right service URL accordingly.
In order to address this inconvenience, you can start a proxy cluster for routing the admin rest requests accordingly. Hence you will have one central entry point for your admin service.
If you already have a proxy cluster, continue reading. If you haven't setup a proxy cluster before, you can follow the instructions to start proxies.
To enable routing functions related admin requests to functions-worker
in a proxy, you can edit the proxy.conf
file to modify the following settings:
functionWorkerWebServiceURL=<pulsar-functions-worker-web-service-url> functionWorkerWebServiceURLTLS=<pulsar-functions-worker-web-service-url>
As described above, you can run Function-worker with brokers, or run it separately. And it is more convenient to run functions-workers along with brokers. However, running functions-workers in a separate cluster provides better resource isolation for running functions in Process
or Thread
mode.
Use which mode for your cases, refer to the following guidelines to determine.
Use the Run-with-Broker
mode in the following cases:
Process
or Thread
mode;Use the Run-separately
mode in the following cases:
Error message: Namespace missing local cluster name in clusters list
Failed to get partitioned topic metadata: org.apache.pulsar.client.api.PulsarClientException$BrokerMetadataException: Namespace missing local cluster name in clusters list: local_cluster=xyz ns=public/functions clusters=[standalone]
The error message prompts when either of the cases occurs:
functionsWorkerEnabled=true
, but the pulsarFunctionsCluster
is not set to the correct cluster in the conf/functions_worker.yaml
file;functionsWorkerEnabled=true
, while brokers in one cluster run well, brokers in the other cluster do not work well.Workaround
If any of these cases happens, follow the instructions below to fix the problem:
Disable Functions Worker by setting functionsWorkerEnabled=false
, and restart brokers.
Get the current clusters list of public/functions
namespace.
bin/pulsar-admin namespaces get-clusters public/functions
bin/pulsar-admin namespaces set-clusters --clusters <existing-clusters>,<new-cluster> public/functions
After setting the cluster successfully, enable functions worker by setting functionsWorkerEnabled=true
.
Set the correct cluster name in pulsarFunctionsCluster
in the conf/functions_worker.yml
file, and restart brokers.