rob-9 opened a new issue, #1034: URL: https://github.com/apache/flink-agents/issues/1034
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description ### Description The Kafka action-state store records action progress so that a recovering job can avoid running completed actions again. Each Flink checkpoint saves a recovery marker with the next Kafka offset to read for every action-state partition. During recovery, `rebuildState()` finds the earliest required offset for each partition and replays the log from there. PR #885 adds opt-in tombstones, which are null-valued Kafka records written for state pruned after a checkpoint completes. Replay treats a tombstone as deletion of that key. On a compacted action-state topic, Kafka may later remove the tombstone and the values it replaced, reclaiming storage and reducing long-term topic growth. Compaction happens asynchronously, so tombstones do not impose a fixed size limit. The recovery trade-off is that an older checkpoint may replay both a completed action-state value and a later tombstone. The tombstone removes the recovered completion state, so the action may run again. The checkpoint that triggered a prune remains usable with respect to that prune, but later pruning may invalidate it once it becomes an older recovery point. Safe physical cleanup therefore needs a clearly defined oldest supported recovery point. The operator and action-state store know the marker saved in an individual checkpoint, but they do not have a complete view of every retained checkpoint, externalized checkpoint, or savepoint that users may still want to restore. For example, suppose C0 needs partition 0 from offset 100 and C1 needs it from offset 150. If the user chooses C1 as the oldest supported recovery point, Kafka may discard records before offset 150. C1 and newer recovery points must still work. An attempt to restore C0 should fail with a clear error instead of silently rebuilding incomplete action state. ### User interaction model - The first version could let users explicitly choose and advance the oldest supported recovery point. The boundary can only move forward. - A later policy could move it automatically, for example by retaining the latest N checkpoints. Users would need to register or pin external savepoints they want the policy to protect. - Once a boundary or policy is configured, cleanup and retries can run without asking for confirmation each time. Without one, coordinated prefix cleanup stays off. The existing opt-in tombstones keep their current recovery trade-off. ### Possible design 1. Keep a durable mapping from each globally completed checkpoint to its effective Kafka recovery marker. For each partition, use the earliest offset required by any subtask so recovery remains correct after rescaling. 2. Limit cleanup to an action-state topic dedicated to one job's recovery history. Kafka's `deleteRecords` removes every record before an offset regardless of which producer wrote it, so independent jobs must not share the topic unless they coordinate the same cleanup boundary. 3. Before deleting anything, durably record the new boundary. After a failover or partial failure, retry the same partition offsets, verify Kafka's resulting beginning offsets, and mark cleanup complete only when every partition reaches its target. 4. Call `deleteRecords` with the effective marker of the oldest supported recovery point. Because the marker is the next offset recovery reads, only records before it may be made unavailable. Kafka may reclaim the underlying storage later. 5. During `rebuildState()`, check that recovery is reading the expected topic and partitions. Each requested offset must be at or after Kafka's beginning offset and no later than its end offset. Otherwise, fail with a clear error that identifies the partition and offsets involved. 6. Disable PR #885's per-key tombstones in this mode, or make them follow the same recovery boundary. Otherwise, a tombstone could still invalidate a checkpoint the policy promises to retain. Suggested by GPT 5.6. ### Open design questions - Where should the checkpoint-marker mapping and cleanup coordinator live, and how should they survive JobManager failover? - How should users register, pin, and release savepoints without racing an ongoing boundary change? - Should the first version support only explicit advancement, leaving count- or time-based policies for later? - How should the system handle upgrades, partition changes, rescaling, or more than one job needing the same recovery history? ### Note This likely needs some discussion before implementation, as it introduces a user-controlled recovery guarantee and coordinates irreversible cleanup across Flink and Kafka. cc @joeyutong @wenjin272 @weiqingy ### Are you willing to submit a PR? - [ ] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
