apache / apache/pulsar

[feat] Pulsar Functions should provide a way to set Pulsar client memory limit

Open
#23,723 4 comments 0 reactions 0 assignees View on GitHub
area/function type/enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.