rob-9 opened a new pull request, #1101: URL: https://github.com/apache/flink-agents/pull/1101
Closes #1034. This branch is stacked on PR #885, which adds opt-in Kafka tombstones for per-key action-state pruning. This change adds checkpoint-aligned prefix cleanup. ### Purpose of change Kafka action-state records can remain essential to Flink checkpoints and savepoints. Deleting a record too early can erase the evidence that an action already completed, allowing recovery to repeat the action and its external side effects. Flink Agents therefore needs a durable agreement about the oldest recovery point that operators still expect to use. This change gives operators a plan/apply workflow for choosing that recovery point and reclaiming the older Kafka prefix. The plan shows the exact partition offsets that become eligible for deletion. Apply records the boundary before asking Kafka to delete data. Recovery then enforces the boundary and gives a clear error for an older checkpoint or savepoint. For example, checkpoint C0 needs partition 0 from offset 100 and C1 needs it from offset 150. Choosing C1 allows Kafka to discard records before offset 150. C1 and newer recovery points remain supported. Restoring C0 stops with an error that names offsets 100 and 150. #### Runtime flow 1. Each checkpoint stores a versioned marker containing the physical Kafka topic ID and next read offset for every action-state partition. 2. The operator runs `KafkaActionStateCleanupTool plan` against the selected checkpoint or savepoint. The tool reads every subtask marker through Flink's State Processor API, takes the minimum offset for each partition, and writes a deterministic JSON plan whose ID comes from its contents. 3. After reviewing the topic identity, partitions, and offsets, the operator runs `apply`. 4. Apply validates the plan and current Kafka state, writes a `COMMITTED` record to a dedicated compacted control topic, and calls `deleteRecords`. 5. Apply verifies the physical topic identity and resulting beginning offsets, then writes `APPLIED`. A `COMMITTED` operation can safely resume after interruption. 6. With `kafkaActionStateCleanupControlTopic` configured, recovery validates its marker and Kafka metadata against the effective committed boundary before replaying the captured offset range. #### Key decisions - The operator selects the boundary because users control the full set of retained checkpoints, externalized checkpoints, and savepoints. - The durable `COMMITTED` record precedes deletion, giving interrupted operations one stable plan to retry. - Plans bind to Kafka's physical topic ID and complete partition set. Apply rechecks that identity around `deleteRecords`, which operates on a topic name that a recreated topic could reuse. - Each partition boundary moves forward or stays equal. This preserves the operator's decision to retire older recovery points. - Legacy map markers support standard recovery. Cleanup requires versioned markers because physical topic identity is essential to safe deletion. #### Related work PR #885 writes tombstones for selected keys. This PR deletes an entire Kafka prefix after the operator advances the oldest supported recovery point. Cleanup mode rejects simultaneous tombstones because each mechanism defines a different recovery guarantee. Issue #1099 fixes typed identity for individual records; the combined replay path must retain its key validation before applying values or tombstones. ### Behavioral Semantics #### Interaction decisions | Control topic | Recovery marker | Tombstones | Result | |---|---|---|---| | Unconfigured | Legacy or versioned | Either setting | Use standard recovery; prefix cleanup stays inactive | | Configured before the first committed plan | Any | Disabled | Stop and request a committed boundary | | Committed boundary | Versioned marker at or above boundary | Disabled | Validate Kafka state and replay the captured range | | Committed boundary | Versioned marker below boundary | Disabled | Stop and report the partition, requested offset, and boundary | | Configured | Legacy marker | Disabled | Stop because physical topic identity and a comparable boundary are absent | | Configured | Any | Enabled | Reject the conflicting recovery policies | #### Behavioral contracts - A plan contains the earliest offset required by any subtask for every partition in the selected recovery point. - Any change to the topic identity, partitions, or offsets produces a different plan ID. - A new boundary equals or advances every partition in the committed boundary. - `COMMITTED` always precedes `deleteRecords`; `APPLIED` follows verification of every resulting beginning offset. - Reapplying an interrupted plan retries the same deletion. - Recovery compares each requested offset with the committed boundary, Kafka beginning offset, and captured end offset before replay. - Apply and recovery verify the physical topic identity and complete partition set. - Observed control records are immutable; a tombstone, changed plan, or status regression raises an error. #### Failure behavior - Invalid JSON, noncanonical fields, missing or duplicate partitions, changed plan contents, and unavailable target offsets stop apply before `COMMITTED`. - A missing or incompatible control topic stops recovery. Apply can create the required single-partition compacted topic. - Kafka client, timeout, deletion, and verification failures propagate. A failure after `COMMITTED` leaves the same plan ready for retry. - Invalid markers, changed Kafka identity, unavailable replay offsets, deserialization failures, and stalled replay stop recovery with contextual errors. ### Tests | Contract | Coverage | |---|---| | Plan minima, deterministic identity, strict JSON, canonical partitions, and tamper rejection | `KafkaActionStateCleanupPlanTest` | | Offset availability, commit-before-delete, monotonic advancement, retry, immutable control records, and topic revalidation | `KafkaActionStateCleanupCoordinatorTest` | | Real Kafka control records, `deleteRecords`, beginning offsets, and boundary reads | `KafkaActionStateCleanupCoordinatorIntegrationTest` | | Marker compatibility, boundary checks, Kafka identity, offset ranges, and captured replay ends | `KafkaActionStateStoreTest` | | Flink union-state reading and CLI plan generation | `KafkaActionStateCleanupToolTest` | | Java and Python configuration parity | Java cleanup tests and `python/flink_agents/api/tests/test_core_options.py` | Verification performed: - Full Java reactor: 1,568 tests passed; 13 tests were skipped. - Python configuration tests: 6 passed. - Ruff 0.11.13, Spotless, and `git diff --check` passed. Remaining integration coverage: the Docker-backed Kafka integration test compiled and awaits a Docker environment. The real Flink savepoint and State Processor path passed locally. <details> <summary>Implementation details for traceability</summary> - The action-state and control topics are distinct and dedicated to one job recovery history. - The control topic has one partition and exactly `cleanup.policy=compact`. - Control records use strict versioned JSON keyed by plan ID and become immutable after observation. - Cleanup checks the physical topic immediately before and after `deleteRecords`. Recovery refreshes metadata before offset reads and after replay. - The recovery consumer uses explicit assignment and seek, disables broker offset commits, and captures replay end offsets before polling. - Testcontainers remains test-scoped. The State Processor API remains provided-scoped and comes from the matching Flink installation. </details> ### API This change adds `kafkaActionStateCleanupControlTopic` to the Java and Python configuration APIs and adds a language-neutral administrative CLI backed by internal Java runtime types. Existing configurations keep coordinated cleanup disabled. Legacy marker maps continue to support standard recovery in that mode. Newly written markers use the versioned, topic-aware format. Cleanup mode adds explicit checks for topic identity, partition identity, boundary position, and offset availability. ### Documentation - [ ] `doc-needed` - [ ] `doc-not-needed` - [x] `doc-included` The deployment and configuration documentation covers the operator workflow, control-topic requirements, recovery guarantees, failure behavior, and tombstone interaction. ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No -- 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]
