DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
We only intend to allow Kafka Streams to maximise use of existing local state when deploying a new version of a Topology. When deploying such upgrades, the entire cluster will still need to be brought down and brought back up together, instead of a rolling restart.
Proposed Changes
There are several different strategies that can be used to resolve or mitigate this problem. We outline 3 proposals with a view to discussing and choosing the most suitable solution.
Note: unless otherwise stated, these proposals are mutually exclusive.
...
New Configuration Properties
| Property | Default | Description |
|---|---|---|
automatic.state.relocation | true | When true, on startup, Kafka Streams will automatically relocate state stores it finds on-disk that are no longer in the correct location. This can occur after changes to a Topology that re-order Sub-Topologies such that Task IDs for existing state are no longer valid. |
Implementation Details
When KafkaStreams#start() is called, if automatic.state.relocation is enabled, Streams will automatically search the configured state directory and move Task state directories to the correct locations according to the current Topology graph.
The "correct" location is defined as: All components of the Path are the same as the current Path, except for the sub-topology ordinal, which is determined from the current Topology graph, by looking for the sub-topology that references the store, by-name.
Example movement
state.dir/2_14/rocksdb/mystore → state.dir/3_14/rocksdb/mystore
Compatibility, Deprecation, and Migration Plan
This change is backwards compatible with previous versions of Kafka Streams. No deprecation or migration is necessary.
The automatic.state.relocation configuration is provided to enable users to disable this behaviour if they need to, but is enabled by default because it should not cause problems for any users.
Moving directories in all modern filesystems is a very cheap operation and O(1), so the performance impact on application start-up, even for application instances with a large number of stateful Tasks, should be minimal.
Rejected Alternatives
Several alternative solutions were explored, but were found to require considerable changes to the internals of Kafka Streams, making them difficult to implement safely.
Alternative 1: De-couple local state directories from Task ID
Local state will be stored under a revised directory structure that no longer includes Task ID in the path.
...
Users that choose not to use this tool to migrate their local state will need to reset their application’s state, to ensure that stale local state does not remain under old paths.
B. Automatically migrate existing local state to new Tasks
The core problem only occurs when a Topology is changed, therefore, this issue only ever manifests during application startup, after an application instance has been modified. Further, this issue is isolated to each instance of the application - issues with local state locations don’t affect the cluster, only the local instance.
If we were to persist a description of the Topology that enables us to automatically detect these breaking changes on start-up, Kafka Streams could automatically migrate local task state from their old TaskIDs to the new ones.
We will store a new file at the root of the state.dir, .store-groups, that stores a mapping of the name of each store changelog topic to an object containing both the name of the StateStore and the Topic Group ordinal it’s assigned to.
When the application starts, after the Topology has been constructed but before joining the ConsumerGroup, the application will load this file if it already exists and compare it to the current Topology.
- For every changelog topic where the ordinal has changed, the existing state for the corresponding store will be relocated under the directory with the new ordinal.
- For every changelog topic that appears in the existing file, but no longer appears in the current Topology, the existing state for the corresponding store will be deleted.
The application will then continue startup as normal, using the previous local state in their new locations to correctly inform Task assignment.
...
Challenges
- Much of the Kafka Streams codebase assumes that TaskId is encoded directly in the state directory path. See
StateDirectory. - Task-wide
.checkpointfiles, which can contain entries for multiple state stores, are currently stored in the task directory. If we decouple the directory from specific Tasks, we will need to either find another way to store these files, or retain the existing task directories exclusively for Task-wide meta-data.
Alternative 2: Change Task ID prefix from an ordinal to a stable hash
The instability of the Task Group ordinal is the reason that existing local state becomes invalidated when the Topology changes. Using a more stable identifier will solve this problem under most circumstances.
...
* There is a small risk of hash collision that will need to be addressed.
Compatibility, Deprecation, and Migration Plan
TBD: This plan will depend on the chosen solution from the Proposed Solution section.
...
.
Challenges
- Still results in state going unnecessarily missing for some sub-topologies (i.e. when a source topic is added to an existing sub-topology).
- Hash collisions that would need to be mitigated, despite being unlikely.
- Massive changes required throughout Kafka Streams codebase.