[feat] Pulsar Functions should provide a way to set Pulsar client memory limit
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before asking
- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Motivation
I haven't found a way to tune the memory limit for the Pulsar client used in Pulsar Functions.
The memory limit is disabled in Pulsar Functions: https://github.com/apache/pulsar/blob/82237d3684fe506bcb6426b3b23f413422e6e4fb/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceUtils.java#L161
There's code for configuring it:
https://github.com/apache/pulsar/blob/05e57dd3a443c5b99c21054c56a1b497455fa867/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java#L106-L168
but it doesn't get called in the JavaInstanceStarter case. The JavaInstanceStarter is used when running Java functions in process / k8s / function mesh runtimes (other than thread runtime):
https://github.com/apache/pulsar/blob/567174f43528c0f7ae917bfb5166213973c62c29/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceStarter.java#L236-L245
`Optional.empty()` is passed to memory limit.
https://github.com/apache/pulsar/blob/05e57dd3a443c5b99c21054c56a1b497455fa867/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntimeFactory.java#L92-L104
It looks like the memory limit solution has been only implemented for the ThreadRuntimeFactory in #9320. When using ThreadRuntimeFactory, it's possible to set the limit in `conf/functions_worker.yml`. This isn't helpful for other runtime types (process, k8s).
### Solution
Add the Pulsar client memory limit configuration to `org.apache.pulsar.common.functions.FunctionConfig`, `org.apache.pulsar.functions.instance.InstanceConfig` and `org.apache.pulsar.functions.worker.WorkerConfig` and make it configurable in the same way as other instance config parameters such as `maxPendingAsyncRequests`.
The default would be configured in `WorkerConfig` for all functions and the function configuration would be in `InstanceConfig` and configurable with `FunctionConfig` in yaml and command line.
It's not fully consistent how configs should be handled. There's also the protobuf `org.apache.pulsar.functions.proto.Function.FunctionDetails` where configs are passed.
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR!
Contributor guide
Research direction
Start by tracing the memory-limit handling in pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceUtils.java, ThreadRuntimeFactory.java, and JavaInstanceStarter.java. Then inspect FunctionConfig, InstanceConfig, WorkerConfig, and FunctionDetails to understand how instance settings are passed. Done means the client memory limit can be configured for the relevant Pulsar Functions runtime types, with configuration behavior consistent with existing instance parameters.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- distributed-systems
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100