PIP-225: Pulsar Functions fetch parameters from local config file.
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
Credits to @tpiperatgod to initialize this PIP, thanks to @freeznet to give feedbacks on this PIP~
### Motivation
The Pulsar functions have the ability to provide the number of instances of the called function (see, for example, [["Get num instances"](https://pulsar.apache.org/docs/2.10.x/window-functions-context/#get-num-instances)](https://pulsar.apache.org/docs/2.10.x/window-functions-context/#get-num-instances) ), but the data that this function currently relies on needs to be passed in via command line arguments and cannot be easily changed dynamically.
In particular, when Pulsar Functions are running in Kubernetes, passing in specific data by updating the command line arguments will trigger a reschedule (restart for pod) of the Pulsar Instance.
Therefore we need to provide a more dynamic way of accessing data (mainly `FunctionDetails`) for functions, and this way can also serve to pass the function's other parameters.
### Goal
- Let the Pulsar functions support reading parameters from the specified configuration file
- Let the Pulsar functions feedback state information more gracefully (no more command line approach)
The current function runtimes that need to support this feature are the Python Runtime, Java Runtime and Golang Runtime.
### API Changes
_No response_
### Implementation
This mechanism is aimed at allowing the function to dynamically fetch the required parameters and data without having to restart the function in order to load the latest parameters or data.
A more appropriate approach is to provide the function with a local config file where the data and parameters are stored. The provider of the data and parameters (usually the initiator of the function, e.g., function-worker) writes the contents of the parameters and the data (FunctionDetails) to a fixed local config file and passes the path of this config file to the function.
```
exec --config-file /path/to/config-file
```
The function determines the value of `--config-file` and determines if the target config file exists, and parses its contents if it does.
The rules for applying parameters or data to the function are as follows.
- Parameters or data passed in on the command line have a higher priority than the corresponding parameters or data in the configuration file
- For mandatory parameters or data, if they are not passed in, an exception is thrown.
- For optional parameters or data, if not passed in, the default value is used
We have two major steps, they are
- Support reading configs from file
- Monitor state config changes
## Support reading configs from file
### Python Runtime
1. Add the argument
`config_file`, default is None, which means you need to get the configuration content from `config_file`
2. The priority of command line arguments is higher than the priority of configuration file arguments during initialization
### Java Runtime
We propose to allow `JavaInstanceStarter` to read from file by:
1. Create a class `JavaInstanceConfiguration extends PulsarConfiguration`. This class will contain all the command line args. Subclassed of `PulsarConfiguration` can be read using `PulsarConfigurationLoader`.
2. In `JavaInstanceStarter`, add another command line args `--config_file`, this arg is the config file location.
3. First , and read the “command line args”, then read the “file configs”. Finally, merge these two with priority “command line args ” > “file configs”, plus required field validation and default value completion.
4. All JCommander configs will be non-required : the actual required validation will be done in a later step through a custom method.
5. If a config field is not provided (by JCommander nor by file), they will be initialized to null value. This way we can distinguish what field are not provided and what are provided but the value is the same as the default value. The default value for each config field is then provided in a later step through a custome method.
`JavaInstanceStarter` is used by all three runtime implementations : `KubernetesRuntime` , `ProcessRuntime` and `ThreadRuntime` . When can then modify the Pulsar Worker to use this new feature to run `JavaInstanceStarter` .
### Golang Runtime
1. The current Golang Runtime reads the configurations from `conf/config.yaml`
2. However, there are structural differences between the configuration items in `conf/config.yaml` and those in Java and Python runtimes, so perhaps we can align their configuration styles in the implementation of this proposal.
An alternative proposal for the coexistence of configuration files and command line arguments is to follow an exclusivity policy, i.e. when a configuration file is applied, then only the configuration is read from the configuration file and no longer care about the arguments passed in via the command line, to be discussed.
## Monitor state config changes
> At present, this requirement mainly arises from Java functions, because the feature of getting the number of function instances is provided in the Window Functions scenario, which is currently supported only by the Java Runtime.
Java instances need to know the parallelism of the current instance. However, this parallelism information is passed in at start-up only, and if the parallelism information changes, we need to change the instance's startup command, which will result in a rescheduling of all instances (process restart or pod restart), which is not desirable.
One way to avoid this reschedule is to avoid changing the instance’s startup command when the parallelism parameter changes. This can be done with our first goal proposal above
In addition, we need to make Pulsar Functions support dynamic fetching the current parallelism information without having to reinitialize each instance due to changes in the number of function instances (possibly through manual changes or auto-scaling based on mechanisms such as [[HPA](https://kubernetes.io/docs/tasks/run-application/horizontal-pod-autoscale/)](https://kubernetes.io/docs/tasks/run-application/horizontal-pod-autoscale/)).
### **What configs should be modifiable at runtime?**
Not every config in `FunctionDetails` should be modifiable at runtime. In fact, some configs are only used during initialization, and these configs (such as `className`, `processingGuarantee`) changes should trigger instance reschedule or process restart. (Discussion here). For such configs, we should not support changing them at runtime without restart.
Configs like `parallelism` should be modifiable at runtime. In fact, instead of defining it as a config entry, it’s more like a context state that is exposed to the Java runtime.
States that need to be modified:
- parallelism
### Implementation Plan: use file to detect config changes
Based on the observation above. Initially our scope will be limited to `FunctionDetails.parallelism` only which used by `Context.getNumInstances()`.
If we find other config options should be supported, we will follow a similar approach:
- Users should provide a config file storing key value config pairs.
- The user provides the location of the config file when starting the instance
And we have two alternatives:
1. Whenever the user queries the config value using context, for example `Context#getNumInstances()`, the runtime should read the file and retrieve the newest value.
2. Use a polling thread to monitor file changes. Whenever the file content is changed, the state of the instance is updated as well.
### Alternatives
Included in the section above.
### Anything else?
Concerns:
- Consistency: If the function relies on all instances see the same config value. For example, the function uses parallelism to compute owned work for each instance, then if two instance sees a different parallelism value, could it cause data inconsistency issues ? (This is like the rebalancing when adding parallelism)
- User code initialization : if the user only uses the config value at initialization to create some objects, then after initialization, even if the config value is updated, the created objects will not be changed.
Contributor guide
Research direction
Start by reading JavaInstanceStarter, PulsarConfiguration, and the Python Runtime configuration handling, then compare them with the existing Go conf/config.yaml approach. The proposal leaves the file format, merge behavior, validation, and change-monitoring alternative unresolved; done would require an agreed design implemented across the three runtimes, including dynamic parallelism for Context#getNumInstances().
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go, java, python
- Domain
- backend, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100