-
Notifications
You must be signed in to change notification settings - Fork 3.7k
Description
Describe the bug
When I enable either the Process or Kubernetes function Runtimes and set the parameter "maxPendingAsyncRequests" to a value other than 1000 in the function worker configuration (e.g. functions_worker.yml), any function instances launched by the worker continue to have maxPendingAsyncRequests = 1000. I'd like to be adjust this parameter to guard against my function running out of memory -- without increasing the function heap size -- due to too many publish operations in flight, as can occur when the output topic's backlog quota has been reached.
Investigation shows that the problem lies in class pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java, method getCmd, which simply does not include the necessary argument for maxPendingAsyncRequests from the InstanceConfig.
To Reproduce
Steps to reproduce the behavior:
- Enable the Process Runtime in the broker:
- Using the pulsar docker image, set the following environment variables when running the pulsar broker:
- PF_functionRuntimeFactoryClassName: org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactory
- PF_containerFactory: process
- Using the pulsar docker image, set the following environment variables when running the pulsar broker:
- Set the maxPendingAsyncRequests property to a value other than 1000
- also in the pulsar broker container set the following environment variable:
- PF_maxPendingAsyncRequests: "250" - Launch a pulsar function
- Look at the top of the pulsar function log, find the dump of the InstanceConfig. You should see "maxPendingAsyncRequests=1000" following the settings for "maxBufferedTuples", "functionAuthenticationSpec", "port", and "clusterName".
Expected behavior
When performing the above steps, I expect to see "maxPendingAsyncRequests=250".
Additional context
(bugfix pull request incoming)