apache / apache/pulsar

function worker doesn't use configurationStoreServers parameter while creating function

Open
#11,261 3 comments 0 reactions 0 assignees View on GitHub
lifecycle/stale type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 22h
Merged PRs (30d)
142

Description

**Describe the bug**
When I try to create a function in my cluster, I have this error :
```
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.167 [pulsar-web-41-5] INFO org.apache.pulsar.functions.worker.rest.api.ComponentImpl - Uploading Function package to fm1/functions/f_java_brain_message_adapter/9fea4fe9-6f33-431a-be10-15f08fe773ac-metrics-1.0-SNAPSHOT-jar-with-dependencies.jar
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.174 [pulsar-web-41-5] INFO org.apache.pulsar.functions.worker.WorkerUtils - Uploading function package to 'fm1/functions/f_java_brain_message_adapter/9fea4fe9-6f33-431a-be10-15f08fe773ac-metrics-1.0-SNAPSHOT-jar-with-dependencies.jar'
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.210 [main-EventThread] INFO org.apache.distributedlog.bk.SimpleLedgerAllocator - Ledger allocator for /pulsar/functions/fm1/functions/f_java_brain_message_adapter/9fea4fe9-6f33-431a-be10-15f08fe773ac-metrics-1.0-SNAPSHOT-jar-with-dependencies.jar//allocation moved version from -1 to 0.
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.219 [main-EventThread] INFO org.apache.distributedlog.BKLogWriteHandler - Initiating Recovery For fm1/functions/f_java_brain_message_adapter/9fea4fe9-6f33-431a-be10-15f08fe773ac-metrics-1.0-SNAPSHOT-jar-with-dependencies.jar: : []
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.221 [pulsar-web-41-5] INFO org.apache.distributedlog.BKLogWriteHandler - Initiating Recovery For fm1/functions/f_java_brain_message_adapter/9fea4fe9-6f33-431a-be10-15f08fe773ac-metrics-1.0-SNAPSHOT-jar-with-dependencies.jar: : []
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.223 [pulsar-web-41-5] INFO org.apache.distributedlog.bk.SimpleLedgerAllocator - Ledger allocator /pulsar/functions/fm1/functions/f_java_brain_message_adapter/9fea4fe9-6f33-431a-be10-15f08fe773ac-metrics-1.0-SNAPSHOT-jar-with-dependencies.jar//allocation moved to phase ALLOCATING : version = 0.
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.225 [pulsar-web-41-5] INFO org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=localhost:2181 sessionTimeout=30000 watcher=org.apache.bookkeeper.zookeeper.ZooKeeperWatcherBase@4d00cb8e
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.225 [pulsar-web-41-5] INFO org.apache.zookeeper.ClientCnxnSocket - jute.maxbuffer value is 10485760 Bytes
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.226 [pulsar-web-41-5] INFO org.apache.zookeeper.ClientCnxn - zookeeper.request.timeout value is 0. feature enabled=false
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.227 [pulsar-web-41-5-SendThread(localhost:2181)] INFO org.apache.zookeeper.ClientCnxn - Opening socket connection to server localhost/127.0.0.1:2181.
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.227 [pulsar-web-41-5-SendThread(localhost:2181)] INFO org.apache.zookeeper.ClientCnxn - SASL config status: Will not attempt to authenticate using SASL (unknown error)
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: 08:52:41.228 [pulsar-web-41-5-SendThread(localhost:2181)] WARN org.apache.zookeeper.ClientCnxn - Session 0x0 for sever localhost/127.0.0.1:2181, Closing socket connection. Attempting reconnect except it is a SessionExpiredException.
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: java.net.ConnectException: Connection refused
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[?:?]
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:777) ~[?:?]
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:344) ~[org.apache.zookeeper-zookeeper-3.6.3.jar:3.6.3]
Jul 08 08:52:41 ip-10-0-2-32 pulsar[10797]: at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1290) [org.apache.zookeeper-zookeeper-3.6.3.jar:3.6.3]
```
I deployed pulsar in cluster mode and Zookeeper service is not installed on broker servers.
ConfigurationStoreServers is set with zookeeper server IP but functions worker try to connect to it with localhost.
```
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: 08:52:36.808 [main] INFO org.apache.pulsar.functions.worker.PulsarWorkerService - Worker Configs: {
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "workerId" : "c-local-fw-10.0.2.32-8080",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "workerHostname" : "10.0.2.32",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "workerPort" : 8080,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "workerPortTls" : 8443,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "authenticateMetricsEndpoint" : true,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "includeStandardPrometheusMetrics" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "jvmGCMetricsLoggerClassName" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "numHttpServerThreads" : 8,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "httpRequestsLimitEnabled" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "httpRequestsMaxPerSecond" : 100.0,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "configurationStoreServers" : "10.0.2.115:2181,10.0.2.15:2181,10.0.2.229:2181",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "zooKeeperSessionTimeoutMillis" : 30000,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "zooKeeperOperationTimeoutSeconds" : 30,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "zooKeeperCacheExpirySeconds" : 300,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "connectorsDirectory" : "./connectors",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "narExtractionDirectory" : "/tmp",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "validateConnectorConfig" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionsDirectory" : "./functions",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionMetadataTopicName" : "metadata",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "useCompactedMetadataTopic" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionWebServiceUrl" : "http://10.0.2.32:8080",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "pulsarServiceUrl" : "pulsar://10.0.2.32:6650",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "pulsarWebServiceUrl" : "http://10.0.2.32:8080",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "clusterCoordinationTopicName" : "coordinate",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "pulsarFunctionsNamespace" : "public/functions",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "pulsarFunctionsCluster" : "local",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "numFunctionPackageReplicas" : 3,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "downloadDirectory" : "./download/pulsar_functions",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "stateStorageServiceUrl" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionAssignmentTopicName" : "assignments",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "schedulerClassName" : "org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "failureCheckFreqMs" : 30000,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "rescheduleTimeoutMs" : 60000,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "rebalanceCheckFreqSec" : -1,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "initialBrokerReconnectMaxRetries" : 60,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "assignmentWriteMaxRetries" : 60,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "instanceLivenessCheckFreqMs" : 30000,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "brokerClientAuthenticationEnabled" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "brokerClientAuthenticationPlugin" : "org.apache.pulsar.client.impl.auth.AuthenticationDisabled",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "brokerClientAuthenticationParameters" : "",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "bookkeeperClientAuthenticationPlugin" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "bookkeeperClientAuthenticationParametersName" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "bookkeeperClientAuthenticationParameters" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "topicCompactionFrequencySec" : 1800,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "tlsEnabled" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "tlsCertificateFilePath" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "tlsKeyFilePath" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "tlsTrustCertsFilePath" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "tlsAllowInsecureConnection" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "tlsRequireTrustedClientCertOnConnect" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "useTls" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "tlsEnableHostnameVerification" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "tlsCertRefreshCheckDurationSec" : 300,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "authenticationEnabled" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "authenticationProviders" : [ ],
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "authorizationEnabled" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "authorizationProvider" : "org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "superUserRoles" : [ ],
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "properties" : { },
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "initializedDlogMetadata" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "brokerClientTrustCertsFilePath" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionRuntimeFactoryClassName" : "org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactory",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionRuntimeFactoryConfigs" : {
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "logDirectory" : "logs/",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "javaInstanceJarLocation" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "pythonInstanceLocation" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "extraFunctionDependenciesDir" : null
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: },
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "secretsProviderConfiguratorClassName" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "secretsProviderConfiguratorConfig" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionInstanceMinResources" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionInstanceMaxResources" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionInstanceResourceGranularities" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionInstanceResourceChangeInLockStep" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionAuthProviderClassName" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "runtimeCustomizerClassName" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "runtimeCustomizerConfig" : { },
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "maxPendingAsyncRequests" : 1000,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "forwardSourceMessageProperty" : true,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionsWorkerServiceNarPackage" : "",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionsWorkerServiceCustomConfigs" : { },
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "exposeAdminClientEnabled" : false,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "threadContainerFactory" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "processContainerFactory" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "kubernetesContainerFactory" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "clientAuthenticationParameters" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "clientAuthenticationPlugin" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionMetadataTopic" : "persistent://public/functions/metadata",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "clusterCoordinationTopic" : "persistent://public/functions/coordinate",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "functionAssignmentTopic" : "persistent://public/functions/assignments",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "tlsTrustChainBytes" : null,
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "workerWebAddress" : "http://10.0.2.32:8080",
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: "workerWebAddressTls" : "https://10.0.2.32:8443"
Jul 08 08:52:36 ip-10-0-2-32 pulsar[10797]: }
```
I ran this command to add my function :
```
pulsar-admin --admin-url http://10.0.2.32:8080 functions create --jar java_functions/metrics-1.0-SNAPSHOT-jar-with-dependencies.jar --name f_java_brain_message_adapter --tenant fm1 --namespace functions --classname com.dopi.metrics.functions.adapters.BrainMessageAdapterFunction --inputs persistent://fm1/topics/dopi-brain-new --parallelism 1 --user-config '{"context-deployment-prefix":"fm1::"}' --retain-ordering
```

**To Reproduce**
Steps to reproduce the behavior:
1. Install pulsar in cluster mode with 2 brokers and 3 zookeepers on separate servers
2. Run function worker with broker
3. Try to add a function

I have the same problem when I configure functions-worker to run separately

**Operating System**
- OS: Debian 10.10 with openjdk-11-jre
- cluster deployed on AWS with terraform and ansible (2 brokers, 3 bookies, 3 zookeeper, 1 proxy)

Contributor guide

Open the contributing guide

Research direction

Start in the function worker entry points named in the report, especially PulsarWorkerService and the function-creation path involving ComponentImpl and WorkerUtils. Trace how configurationStoreServers is passed during function creation and compare it with the localhost ZooKeeper connection shown in the logs. Done means function creation uses the configured ZooKeeper servers in cluster mode and a regression test covers that configuration.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
distributed-systems
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.