Not able to run pulsar function locally

[raunakagrawal47]

I have tried everything to package and run my python function locally.
I tried to run the pulsar standalone on my laptop (v. 2.9.1) as well as tried to run pulsar in docker (latest image).

Here is the repo: https://github.com/raunakagrawal47/pulsar-test

I am trying to run a basic python function with some dependencies attached to it.
Steps I followed to run pulsar locally:
Generate whl file for all required dependencies: pip3 download --only-binary :all: -r requirements.txt -d deps
Zip the contents of folder: zip -r format-phone-number.zip . -x test/**\* -x venv/**\* -x .idea/**\*
Run the pulsar function: bin/pulsar-admin functions localrun --tenant public --namespace default --py format-phone-number.zip --classname TestEtl.TestEtl --inputs persistent://public/default/in --output persistent://public/default/out

I tried to run pulsar function locally on laptop by running pulsar standalone as well as tried to run by copying the zip file to a pulsar docker image using:

docker cp format-phone-number.zip bbcba9a3f42b:/pulsar
docker exec -it bbcba9a3f42b /bin/bash

When running, I am getting below error:

2022-11-20T14:58:51,194+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactory - Java instance jar location is not defined, using the location defined in system environment : /pulsar/instances/java-instance.jar
2022-11-20T14:58:51,197+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactory - Python instance file location is not defined using the location defined in system environment : /pulsar/instances/python-instance/python_instance_main.py
2022-11-20T14:58:51,197+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactory - No extra dependencies location is defined in either function worker config or system environment
2022-11-20T14:58:51,265+0000 [main] INFO org.apache.pulsar.functions.runtime.RuntimeSpawner - public/default/TestEtl-0 RuntimeSpawner starting function
2022-11-20T14:58:51,266+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - Creating function log directory /pulsar/logs/functions/public/default/TestEtl
2022-11-20T14:58:51,267+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - Created or found function log directory /pulsar/logs/functions/public/default/TestEtl
2022-11-20T14:58:51,268+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - ProcessBuilder starting the process with args python /pulsar/instances/python-instance/python_instance_main.py --py format-phone-number.zip --logging_directory /pulsar/logs/functions --logging_file TestEtl --logging_config_file /pulsar/conf/functions-logging/logging_config.ini --instance_id 0 --function_id 44d3a24d-a564-45ed-a817-9b55d3212351 --function_version 98cd871d-14ee-40d6-8272-fdc411b834d3 --function_details '{"tenant":"public","namespace":"default","name":"TestEtl","className":"TestEtl.TestEtl","runtime":"PYTHON","autoAck":true,"parallelism":1,"source":{"inputSpecs":{"persistent://public/default/in":{}},"cleanupSubscription":true},"sink":{"topic":"persistent://public/default/out","forwardSourceMessageProperty":true},"resources":{"cpu":1.0,"ram":"1073741824","disk":"10737418240"},"componentType":"FUNCTION"}' --pulsar_serviceurl pulsar://localhost:6650 --use_tls false --tls_allow_insecure false --hostname_verification_enabled false --max_buffered_tuples 1024 --port 33167 --metrics_port 42809 --expected_healthcheck_interval 30 --secrets_provider secretsprovider.ClearTextSecretsProvider --cluster_name local
2022-11-20T14:58:51,280+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - Started process successfully
2022-11-20 14:58:51.526 INFO [140611343988544] Client:88 | Subscribing on Topic :persistent://public/default/in
2022-11-20 14:58:51.526 INFO [140611343988544] ClientConnection:189 | [<none> -> pulsar://localhost:6650] Create ClientConnection, timeout=10000
2022-11-20 14:58:51.526 INFO [140611343988544] ConnectionPool:96 | Created connection for pulsar://localhost:6650
2022-11-20 14:58:51.526 INFO [140611281073920] ExecutorService:41 | Run io_service in a single thread
2022-11-20 14:58:51.527 INFO [140611281073920] ClientConnection:375 | [127.0.0.1:59164 -> 127.0.0.1:6650] Connected to broker
2022-11-20 14:58:51.535 INFO [140611281073920] HandlerBase:64 | [persistent://public/default/in, public/default/TestEtl, 0] Getting connection from pool
2022-11-20 14:58:51.535 INFO [140611055253248] ExecutorService:41 | Run io_service in a single thread
2022-11-20 14:58:51.540 INFO  [140611281073920] ConsumerImpl:224 | [persistent://public/default/in, public/default/TestEtl, 0] Created consumer on broker [127.0.0.1:59164 -> 127.0.0.1:6650] 
2022-11-20T14:59:21,662+0000 [function-timer-thread-1-1] ERROR org.apache.pulsar.functions.runtime.process.ProcessRuntime - Health check failed for TestEtl-0
java.util.concurrent.ExecutionException: io.grpc.StatusRuntimeException: UNAVAILABLE: io exception
 at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395) ~[?:?]
 at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999) ~[?:?]
 at org.apache.pulsar.functions.runtime.process.ProcessRuntime.lambda$start$1(ProcessRuntime.java:184) ~[org.apache.pulsar-pulsar-functions-runtime-2.10.1.jar:2.10.1]
 at org.apache.pulsar.common.util.Runnables$CatchingAndLoggingRunnable.run(Runnables.java:54) [org.apache.pulsar-pulsar-common-2.10.1.jar:2.10.1]
 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
 at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) [?:?]
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) [?:?]
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
 at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.77.Final.jar:4.1.77.Final]
 at java.lang.Thread.run(Thread.java:829) [?:?]
Caused by: io.grpc.StatusRuntimeException: UNAVAILABLE: io exception
 at io.grpc.Status.asRuntimeException(Status.java:535) ~[io.grpc-grpc-api-1.45.1.jar:1.45.1]
 at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:533) ~[io.grpc-grpc-stub-1.45.1.jar:1.45.1]
 at io.grpc.internal.DelayedClientCall$DelayedListener$3.run(DelayedClientCall.java:463) ~[io.grpc-grpc-core-1.45.1.jar:1.45.1]
 at io.grpc.internal.DelayedClientCall$DelayedListener.delayOrExecute(DelayedClientCall.java:427) ~[io.grpc-grpc-core-1.45.1.jar:1.45.1]
 at io.grpc.internal.DelayedClientCall$DelayedListener.onClose(DelayedClientCall.java:460) ~[io.grpc-grpc-core-1.45.1.jar:1.45.1]
 at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:562) ~[io.grpc-grpc-core-1.45.1.jar:1.45.1]
 at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70) ~[io.grpc-grpc-core-1.45.1.jar:1.45.1]
 at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:743) ~[io.grpc-grpc-core-1.45.1.jar:1.45.1]
 at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:722) ~[io.grpc-grpc-core-1.45.1.jar:1.45.1]
 at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37) ~[io.grpc-grpc-core-1.45.1.jar:1.45.1]
 at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133) ~[io.grpc-grpc-core-1.45.1.jar:1.45.1]
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?]
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]
 ... 1 more
Caused by: io.grpc.netty.shaded.io.netty.channel.AbstractChannel$AnnotatedConnectException: finishConnect(..) failed: Connection refused: /127.0.0.1:33167
Caused by: java.net.ConnectException: finishConnect(..) failed: Connection refused
 at io.grpc.netty.shaded.io.netty.channel.unix.Errors.newConnectException0(Errors.java:155) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 at io.grpc.netty.shaded.io.netty.channel.unix.Errors.handleConnectErrno(Errors.java:128) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 at io.grpc.netty.shaded.io.netty.channel.unix.Socket.finishConnect(Socket.java:320) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 at io.grpc.netty.shaded.io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.doFinishConnect(AbstractEpollChannel.java:710) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 at io.grpc.netty.shaded.io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.finishConnect(AbstractEpollChannel.java:687) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 at io.grpc.netty.shaded.io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.epollOutReady(AbstractEpollChannel.java:567) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 at io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:470) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 at io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 at io.grpc.netty.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 at io.grpc.netty.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 at io.grpc.netty.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.grpc-grpc-netty-shaded-1.45.1.jar:1.45.1]
 ... 1 more
2022-11-20T14:59:21,676+0000 [function-timer-thread-1-1] ERROR org.apache.pulsar.functions.runtime.process.ProcessRuntime - Extracted Process death exception
java.lang.RuntimeException: 
 at org.apache.pulsar.functions.runtime.process.ProcessRuntime.tryExtractingDeathException(ProcessRuntime.java:400) ~[org.apache.pulsar-pulsar-functions-runtime-2.10.1.jar:2.10.1]
 at org.apache.pulsar.functions.runtime.process.ProcessRuntime.isAlive(ProcessRuntime.java:387) ~[org.apache.pulsar-pulsar-functions-runtime-2.10.1.jar:2.10.1]
 at org.apache.pulsar.functions.runtime.RuntimeSpawner.lambda$start$0(RuntimeSpawner.java:88) ~[org.apache.pulsar-pulsar-functions-runtime-2.10.1.jar:2.10.1]
 at org.apache.pulsar.common.util.Runnables$CatchingAndLoggingRunnable.run(Runnables.java:54) [org.apache.pulsar-pulsar-common-2.10.1.jar:2.10.1]
 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
 at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) [?:?]
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) [?:?]
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
 at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.77.Final.jar:4.1.77.Final]
 at java.lang.Thread.run(Thread.java:829) [?:?]
2022-11-20T14:59:21,677+0000 [function-timer-thread-1-1] ERROR org.apache.pulsar.functions.runtime.RuntimeSpawner - public/default/TestEtl Function Container is dead with following exception. Restarting.
java.lang.RuntimeException: 
 at org.apache.pulsar.functions.runtime.process.ProcessRuntime.tryExtractingDeathException(ProcessRuntime.java:400) ~[org.apache.pulsar-pulsar-functions-runtime-2.10.1.jar:2.10.1]
 at org.apache.pulsar.functions.runtime.process.ProcessRuntime.isAlive(ProcessRuntime.java:387) ~[org.apache.pulsar-pulsar-functions-runtime-2.10.1.jar:2.10.1]
 at org.apache.pulsar.functions.runtime.RuntimeSpawner.lambda$start$0(RuntimeSpawner.java:88) ~[org.apache.pulsar-pulsar-functions-runtime-2.10.1.jar:2.10.1]
 at org.apache.pulsar.common.util.Runnables$CatchingAndLoggingRunnable.run(Runnables.java:54) [org.apache.pulsar-pulsar-common-2.10.1.jar:2.10.1]
 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
 at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) [?:?]
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) [?:?]
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
 at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.77.Final.jar:4.1.77.Final]
 at java.lang.Thread.run(Thread.java:829) [?:?]
2022-11-20T14:59:21,684+0000 [function-timer-thread-1-1] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - Creating function log directory /pulsar/logs/functions/public/default/TestEtl
2022-11-20T14:59:21,684+0000 [function-timer-thread-1-1] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - Created or found function log directory /pulsar/logs/functions/public/default/TestEtl
2022-11-20T14:59:21,684+0000 [function-timer-thread-1-1] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - ProcessBuilder starting the process with args python /pulsar/instances/python-instance/python_instance_main.py --py format-phone-number.zip --logging_directory /pulsar/logs/functions --logging_file TestEtl --logging_config_file /pulsar/conf/functions-logging/logging_config.ini --instance_id 0 --function_id 44d3a24d-a564-45ed-a817-9b55d3212351 --function_version 98cd871d-14ee-40d6-8272-fdc411b834d3 --function_details '{"tenant":"public","namespace":"default","name":"TestEtl","className":"TestEtl.TestEtl","runtime":"PYTHON","autoAck":true,"parallelism":1,"source":{"inputSpecs":{"persistent://public/default/in":{}},"cleanupSubscription":true},"sink":{"topic":"persistent://public/default/out","forwardSourceMessageProperty":true},"resources":{"cpu":1.0,"ram":"1073741824","disk":"10737418240"},"componentType":"FUNCTION"}' --pulsar_serviceurl pulsar://localhost:6650 --use_tls false --tls_allow_insecure false --hostname_verification_enabled false --max_buffered_tuples 1024 --port 33167 --metrics_port 42809 --expected_healthcheck_interval 30 --secrets_provider secretsprovider.ClearTextSecretsProvider --cluster_name local
2022-11-20T14:59:21,685+0000 [Timer-0] INFO org.apache.pulsar.functions.LocalRunner - {
 "failureException": "UNAVAILABLE: io exception",
 "instanceId": "0"
}

I tried looking at logs inside: pulsar/logs/functions/public/default/TestEtl:
Here is the output:

[2022-11-20 15:25:25 +0000] [ERROR] log.py: Could not import User Function Module TestEtl.TestEtl
[2022-11-20 15:25:55 +0000] [INFO] python_instance_main.py: Starting Python instance with Namespace(client_auth_params=None, client_auth_plugin=None, cluster_name='local', dependency_repository=None, expected_healthcheck_interval=30, extra_dependency_repository=None, function_details='{"tenant":"public","namespace":"default","name":"TestEtl","className":"TestEtl.TestEtl","runtime":"PYTHON","autoAck":true,"parallelism":1,"source":{"inputSpecs":{"persistent://public/default/in":{}},"cleanupSubscription":true},"sink":{"topic":"persistent://public/default/out","forwardSourceMessageProperty":true},"resources":{"cpu":1.0,"ram":"1073741824","disk":"10737418240"},"componentType":"FUNCTION"}', function_id='73da1f7a-97bc-4274-942a-377dd0e4727e', function_version='575e5fd1-4dce-4f8c-81d7-62e02c68ec99', hostname_verification_enabled='false', install_usercode_dependencies=None, instance_id='0', logging_config_file='/pulsar/conf/functions-logging/logging_config.ini', logging_directory='/pulsar/logs/functions', logging_file='TestEtl', max_buffered_tuples='1024', metrics_port=35879, port=34267, pulsar_serviceurl='pulsar://localhost:6650', py='format-phone-number.zip', secrets_provider='secretsprovider.ClearTextSecretsProvider', secrets_provider_config=None, state_storage_serviceurl=None, tls_allow_insecure_connection='false', tls_trust_cert_path=None, use_tls='false')
[2022-11-20 15:25:55 +0000] [INFO] util.py: Failed to import class TestEtl.TestEtl from path 
[2022-11-20 15:25:55 +0000] [INFO] util.py: No module named 'TestEtl'
Traceback (most recent call last):
 File "/pulsar/instances/python-instance/util.py", line 41, in import_class
 return import_class_from_path(from_path, full_class_name)
 File "/pulsar/instances/python-instance/util.py", line 64, in import_class_from_path
 mod = importlib.import_module(classname_path)
 File "/usr/lib/python3.8/importlib/__init__.py", line 127, in import_module
 return _bootstrap._gcd_import(name[level:], package, level)
 File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
 File "<frozen importlib._bootstrap>", line 991, in _find_and_load
 File "<frozen importlib._bootstrap>", line 973, in _find_and_load_unlocked
ModuleNotFoundError: No module named 'TestEtl'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
 File "/pulsar/instances/python-instance/util.py", line 46, in import_class
 return import_class_from_path(api_dir, full_class_name)
 File "/pulsar/instances/python-instance/util.py", line 64, in import_class_from_path
 mod = importlib.import_module(classname_path)
 File "/usr/lib/python3.8/importlib/__init__.py", line 127, in import_module
 return _bootstrap._gcd_import(name[level:], package, level)
 File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
 File "<frozen importlib._bootstrap>", line 991, in _find_and_load
 File "<frozen importlib._bootstrap>", line 973, in _find_and_load_unlocked
ModuleNotFoundError: No module named 'TestEtl'
[2022-11-20 15:25:55 +0000] [CRITICAL] python_instance.py: Could not import User Function Module TestEtl.TestEtl
[2022-11-20 15:25:55 +0000] [ERROR] log.py: Traceback (most recent call last):
[2022-11-20 15:25:55 +0000] [ERROR] log.py:  File "/pulsar/instances/python-instance/python_instance_main.py", line 218, in <module>
[2022-11-20 15:25:55 +0000] [ERROR] log.py: main()
[2022-11-20 15:25:55 +0000] [ERROR] log.py:  File "/pulsar/instances/python-instance/python_instance_main.py", line 199, in main
[2022-11-20 15:25:55 +0000] [ERROR] log.py: pyinstance.run()
[2022-11-20 15:25:55 +0000] [ERROR] log.py:  File "/pulsar/instances/python-instance/python_instance.py", line 199, in run
[2022-11-20 15:25:55 +0000] [ERROR] log.py: raise NameError("Could not import User Function Module %s" % self.instance_config.function_details.className)
[2022-11-20 15:25:55 +0000] [ERROR] log.py: NameError
[2022-11-20 15:25:55 +0000] [ERROR] log.py: :
[2022-11-20 15:25:55 +0000] [ERROR] log.py: Could not import User Function Module TestEtl.TestEtl

Please help me out with this. cc: @Jason918

[Raunak-Agrawal]

@Jason918 @michaeljmarshall can you please help with this? Basically I am trying to run pulsar function with python source code along with some of its external dependencies. I tried to build wheel file and use pulsar-admin create command to upload the function to the pulsar container, but it fails with some module not found error.
I am using Mac os Monetory. Could you help me with the steps ?