functions_worker.yml

#
# 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.
#

####################
# Worker Settings
####################

workerId: standalone
workerHostname: localhost
# Set to null to disable the non-TLS service port.
workerPort: 6750
workerPortTls: 6751

# The Configuration metadata store url
# Examples:
# * zk:my-zk-1:2181,my-zk-2:2181,my-zk-3:2181
# * my-zk-1:2181,my-zk-2:2181,my-zk-3:2181 (will default to ZooKeeper when the schema is not specified)
# * zk:my-zk-1:2181,my-zk-2:2181,my-zk-3:2181/my-chroot-path (to add a ZK chroot path)
configurationMetadataStoreUrl: zk:localhost:2181

# Metadata store session timeout in milliseconds
metadataStoreSessionTimeoutMillis: 30000

# Metadata store operation timeout in seconds
metadataStoreOperationTimeoutSeconds: 30

# Metadata store cache expiry time in seconds
metadataStoreCacheExpirySeconds: 300

################################
# Function package management
################################

numFunctionPackageReplicas: 1
downloadDirectory: download/pulsar_functions
#Enable Functions Worker to use packageManagement Service to store package.
functionsWorkerEnablePackageManagement: false
# Classname of Pluggable JVM GC metrics logger that can log GC specific metrics
# jvmGCMetricsLoggerClassName:

################################
# Function Metrics
################################

# Whether the '/metrics' endpoint requires authentication. Defaults to true
authenticateMetricsEndpoint: true

#################################################################
# Function metadata management (assignment, scheduling, and etc)
#################################################################

# Configure the pulsar client used by function metadata management
#
# points
# Whether to enable TLS when clients connect to broker
useTls: false
# For TLS:
# brokerServiceUrl=pulsar+ssl://localhost:6651/
pulsarServiceUrl: pulsar://localhost:6650
# For TLS:
# webServiceUrl=https://localhost:8443/
pulsarWebServiceUrl: http://localhost:8080

############################################
# security settings for pulsar broker client
############################################
# The path to trusted certificates used by the Pulsar client to authenticate with Pulsar brokers
# brokerClientTrustCertsFilePath:
# Whether to enable the broker client authentication used by function workers to talk to brokers
# brokerClientAuthenticationEnabled: false
# the authentication plugin to be used by the pulsar client used in worker service
# brokerClientAuthenticationPlugin:
# the authentication parameter to be used by the pulsar client used in worker service
# brokerClientAuthenticationParameters:

# Bookie Authentication
#
# Authentication plugin to use when connecting to bookies
# bookkeeperClientAuthenticationPlugin:
# BookKeeper auth plugin implementation specifics parameters name and values
# bookkeeperClientAuthenticationParametersName:
# Parameters for bookkeeper auth plugin
# bookkeeperClientAuthenticationParameters:


# pulsar topics used for function metadata management

pulsarFunctionsNamespace: public/functions
pulsarFunctionsCluster: standalone
functionMetadataTopicName: metadata
clusterCoordinationTopicName: coordinate
# Should the metadata topic be compacted? Defaulting it to false for backwards compatibility
useCompactedMetadataTopic: false
# Number of threads to use for HTTP requests processing. Default is set to 8
numHttpServerThreads: 8

# function assignment and scheduler

schedulerClassName: "org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler"
functionAssignmentTopicName: "assignments"
failureCheckFreqMs: 30000
rescheduleTimeoutMs: 60000
# frequency at which to check if cluster needs rebalancing (set to -1 to disable)
rebalanceCheckFreqSec: -1
initialBrokerReconnectMaxRetries: 60
assignmentWriteMaxRetries: 60
instanceLivenessCheckFreqMs: 30000
# Frequency how often worker performs compaction on function-topics
topicCompactionFrequencySec: 1800
# should the function context have pulsar admin client exposed? Default is disabled.
exposeAdminClientEnabled: false


###############################
# Function Runtime Management
###############################

#### Process Runtime  ####
# Pulsar function instances are launched as processes

functionRuntimeFactoryClassName: org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactory
functionRuntimeFactoryConfigs:
    # location of log files for functions
    logDirectory: logs/
    # change the jar location only when you put the java instance jar in a different location
    javaInstanceJarLocation:
    # change the python instance location only when you put the python instance jar in a different location
    pythonInstanceLocation:
    # change the extra dependencies location:
    extraFunctionDependenciesDir:

#### Additional JVM tuning (only process and Kubernetes runtime) ####
#   This arguments will be added to the command line execution of 'java'
#additionalJavaRuntimeArguments: ['-XX:+ExitOnOutOfMemoryError']

#### Thread Runtime ####
# Pulsar function instances are run as threads

#functionRuntimeFactoryClassName: org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory
#functionRuntimeFactoryConfigs:
#  # thread group name
#  threadGroupName: "Thread Function Container Group"
## Set the pulsar client memory limit
#pulsarClientMemoryLimit
#    # the max memory in bytes the pulsar client can use
#    absoluteValue:
#    # the max memory the pulsar client can use as a percentage of max direct memory set for JVM
#    percentOfMaxDirectMemory:

#### Kubernetes Runtime ####
# Pulsar function are deployed to Kubernetes

# Upload the builtin sources/sinks/functions to BookKeeper.
# True by default.
# uploadBuiltinSinksSources: true
#functionRuntimeFactoryClassName: org.apache.pulsar.functions.runtime.kubernetes.KubernetesRuntimeFactory
#functionRuntimeFactoryConfigs:
#    # uri to kubernetes cluster, leave it to empty and it will use the kubernetes settings in function worker
#    k8Uri:
#    # the kubernetes namespace to run the function instances. it is `default`, if this setting is left to be empty
#    jobNamespace:
#    # The Kubernetes pod name to run the function instances. It is set to
#    # `pf-<tenant>-<namespace>-<function_name>-<random_uuid(8)>` if this setting is left to be empty
#    jobName:
#    # the docker image to run function instance. by default it is `apachepulsar/pulsar`
#    pulsarDockerImageName:
#    # the docker image to run function instance according to different configurations provided by users.
#    # By default it is `apachepulsar/pulsar`.
#    # e.g:
#    # functionDockerImages:
#    #   JAVA: JAVA_IMAGE_NAME
#    #   PYTHON: PYTHON_IMAGE_NAME
#    #   GO: GO_IMAGE_NAME
#    functionDockerImages:
#    # "The image pull policy for image used to run function instance. By default it is `IfNotPresent`
#    imagePullPolicy: IfNotPresent
#    # the root directory of pulsar home directory in `pulsarDockerImageName`. by default it is `/pulsar`.
#    # if you are using your own built image in `pulsarDockerImageName`, you need to set this setting accordingly
#    pulsarRootDir:
#    # The config admin CLI allows users to customize the configuration of the admin cli tool, such as:
#    # `/bin/pulsar-admin and /bin/pulsarctl`. By default it is `/bin/pulsar-admin`. If you want to use `pulsarctl`
#    # you need to set this setting accordingly
#    configAdminCLI:
#    # this setting only takes effects if `k8Uri` is set to null. if your function worker is running as a k8 pod,
#    # setting this to true is let function worker to submit functions to the same k8s cluster as function worker
#    # is running. setting this to false if your function worker is not running as a k8 pod.
#    submittingInsidePod: false
#    # setting the pulsar service url that pulsar function should use to connect to pulsar
#    # if it is not set, it will use the pulsar service url configured in worker service
#    pulsarServiceUrl:
#    # setting the pulsar admin url that pulsar function should use to connect to pulsar
#    # if it is not set, it will use the pulsar admin url configured in worker service
#    pulsarAdminUrl:
#    # The flag indicates to install user code dependencies. (applied to python package)
#    installUserCodeDependencies:
#    # The repository that pulsar functions use to download python dependencies
#    pythonDependencyRepository:
#    # The repository that pulsar functions use to download extra python dependencies
#    pythonExtraDependencyRepository:
#    # the custom labels that function worker uses to select the nodes for pods
#    customLabels:
#    # The expected metrics collection interval, in seconds
#    expectedMetricsCollectionInterval: 30
#    # Kubernetes Runtime will periodically checkback on
#    # this configMap if defined and if there are any changes
#    # to the kubernetes specific stuff, we apply those changes
#    changeConfigMap:
#    # The namespace for storing change config map
#    changeConfigMapNamespace:
#    # The ratio cpu request and cpu limit to be set for a function/source/sink.
#    # The formula for cpu request is cpuRequest = userRequestCpu / cpuOverCommitRatio
#    cpuOverCommitRatio: 1.0
#    # The ratio memory request and memory limit to be set for a function/source/sink.
#    # The formula for memory request is memoryRequest = userRequestMemory / memoryOverCommitRatio
#    memoryOverCommitRatio: 1.0
#    # The port inside the function pod which is used by the worker to communicate with the pod
#    grpcPort: 9093
#    # The port inside the function pod on which prometheus metrics are exposed
#    # An empty value disables prometheus metrics.
#    metricsPort: 9094
#    # The directory inside the function pod where nar packages will be extracted
#    narExtractionDirectory:
#    # The classpath where function instance files stored
#    functionInstanceClassPath:
#    # the directory for dropping extra function dependencies
#    # if it is not an absolute path, it is relative to `pulsarRootDir`
#    extraFunctionDependenciesDir:
#    # Additional memory padding added on top of the memory requested by the function per on a per instance basis
#    percentMemoryPadding: 10
#    # The duration in seconds before the StatefulSet deleted on function stop/restart.
#    # Value must be non-negative integer. The value zero indicates delete immediately.
#    # Default is 5 seconds.
#    gracePeriodSeconds: 5

## A set of the minimum amount of resources functions must request.
## Support for this depends on function runtime.
## Only kubernetes runtime currently supports this.
# functionInstanceMinResources:
#   cpu: 1
#   ram: 1073741824
#   disk: 10737418240

## A set of the maximum amount of resources functions may request.
## Support for this depends on function runtime.
## Only kubernetes runtime currently supports this.
# functionInstanceMaxResources:
#   cpu: 16
#   ram: 17179869184
#   disk: 107374182400

## Granularities of requested resources.
## If the granularity of any type of resource is set,
## the requested resource of the type must be a multiple of the granularity.
#functionInstanceResourceGranularities:
#    cpu: 1.0
#    ram: 1073741824
#    disk: 10737418240

## If this configuration is set to be true, the amount of requested resources of all type of resources
## that have the granularity set must be the same multiples of their granularities.
#functionInstanceResourceChangeInLockStep: true

## The full class-name of an instance of RuntimeCustomizer.
## This class receives the customRuntimeOptions string and can customize details of how the runtime operates.
#runtimeCustomizerClassName: "org.apache.pulsar.functions.runtime.kubernetes.KubernetesManifestCustomizer"

## This config will pass to RuntimeCustomizer's initialize function to do initializing.
#runtimeCustomizerConfig:
#   extractLabels:
#       extraLabel: value
#   extraAnnotations:
#       extraAnnotation: value
#   nodeSelectorLabels:
#       customLabel: value
#   jobNamespace: namespace
#   tolerations:
#   - key: custom-key
#     value: value
#     effect: NoSchedule

## Config admin CLI
#configAdminCLI:

############################################
# security settings for worker service
############################################

# Enforce authentication
authenticationEnabled: false
# Enforce authorization on accessing functions api
authorizationEnabled: false
# Set of authentication provider name list, which is a list of class names
authenticationProviders:
# Authorization provider fully qualified class-name
authorizationProvider: org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
# Set of role names that are treated as "super-user", meaning they will be able to access any admin-api
superUserRoles:
# Set of role names that are treated as "proxy" roles. These are the roles that can supply the originalPrincipal.
proxyRoles:

#### tls configuration for worker service
# Enable TLS
tlsEnabled: false
# Path for the TLS certificate file
tlsCertificateFilePath:
# Path for the TLS private key file
tlsKeyFilePath:
# Path for the trusted TLS certificate file
tlsTrustCertsFilePath:
# Accept untrusted TLS certificate from client
tlsAllowInsecureConnection: false
# Whether server hostname must match the common name of the certificate
tlsEnableHostnameVerification: false
# Tls cert refresh duration in seconds (set 0 to check on every new connection)
tlsCertRefreshCheckDurationSec: 300
# Whether client certificates are required for TLS. Connections are rejected if the client
# certificate isn't trusted.
tlsRequireTrustedClientCertOnConnect: false

### --- TLS config variables --- ###
## Note that some of the above TLS configs also apply to the KeyStore TLS configuration.

# Specify the TLS provider for the web service: SunJSSE, Conscrypt and etc.
tlsProvider: Conscrypt

# Enable TLS with KeyStore type configuration in function worker.
tlsEnabledWithKeyStore: false

# TLS KeyStore type configuration in function worker: JKS, PKCS12
tlsKeyStoreType: JKS

# TLS KeyStore path in function worker
tlsKeyStore:

# TLS KeyStore password for function worker
tlsKeyStorePassword:

# TLS TrustStore type configuration in function worker: JKS, PKCS12
tlsTrustStoreType: JKS

# TLS TrustStore path in function worker
tlsTrustStore:

# TLS TrustStore password in function worker, default value is empty password
tlsTrustStorePassword:

# Specify the tls protocols the function worker's web service will use to negotiate during TLS handshake
# (a comma-separated list of protocol names).
# Examples:- [TLSv1.3, TLSv1.2]
webServiceTlsProtocols:

# Specify the tls cipher the function worker will use to negotiate during TLS Handshake
# (a comma-separated list of ciphers).
# Examples:- [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256]
webServiceTlsCiphers:

### --- SASL Authentication Provider --- ###

# This is a regexp, which limits the range of possible ids which can connect to the Broker using SASL.
# Default value: `SaslConstants.JAAS_CLIENT_ALLOWED_IDS_DEFAULT`, which is ".*pulsar.*",
# so only clients whose id contains 'pulsar' are allowed to connect.
saslJaasClientAllowedIds: .*pulsar.*

# Service Principal, for login context name.
# Default value `SaslConstants.JAAS_DEFAULT_BROKER_SECTION_NAME`, which is "PulsarFunction".
saslJaasServerSectionName: PulsarFunction

# Path to file containing the secret to be used to SaslRoleTokenSigner
# The Path can be specified like:
# saslJaasServerRoleTokenSignerSecretPath=file:///my/saslRoleTokenSignerSecret.key
saslJaasServerRoleTokenSignerSecretPath:

########################
# State Management
########################

# the service url points to bookkeeper table service
# stateStorageServiceUrl: bk://localhost:4181

########################
# Connectors
########################

connectorsDirectory: ./connectors
functionsDirectory: ./functions

# Should connector config be validated during submission
validateConnectorConfig: false

# Whether to initialize distributed log metadata by runtime.
# If it is set to true, you must ensure that it has been initialized by "bin/pulsar initialize-cluster-metadata" command.
initializedDlogMetadata: false

# Whether to ignore unknown properties when deserializing the connector configuration.
# After upgrading a connector to a new version with a new configuration, the new configuration may not be compatible with the old connector.
# In case of rollback, it's required to also rollback the connector configuration.
# Ignoring unknown fields makes possible to keep the new configuration and only rollback the connector.
ignoreUnknownConfigFields: false

###########################
# Arbitrary Configuration
###########################
# When a configuration parameter is not explicitly named in the WorkerConfig class, it is only accessible from the
# properties map. This map can be configured by supplying values to the properties map in this config file.

# Configure the DLog bookkeeper client by prefixing configurations with "bookkeeper_". Because these are arbitrary, they
# must be added to the properties map to get correctly applied. This configuration applies to the Dlog bookkeeper client
# in both the standalone function workers and function workers initialized in the broker.

# You can add extra configuration options for the Pulsar Client and the Pulsar Admin Client
# by prefixing them with "brokerClient_". These configurations are applied after hard coded configuration
# and before the above brokerClient configurations named above.

## For example, when using the token authentication provider (AuthenticationProviderToken), you must configure several
## custom configurations. Here is a sample for configuring one of the necessary configs:
#properties:
#    tokenPublicKey: "file:///path/to/my/key"
#    tokenPublicAlg: "RSA256"

### --- Deprecated settings --- ###
configurationStoreServers: localhost:2181

# ZooKeeper session timeout in milliseconds
# Deprecated: use metadataStoreSessionTimeoutMillis
zooKeeperSessionTimeoutMillis: -1

# ZooKeeper operation timeout in seconds
# Deprecated: use metadataStoreOperationTimeoutSeconds
zooKeeperOperationTimeoutSeconds: -1