[runtime][api][python] Add checkpoint-aligned Kafka action-state cleanup

[rob-9]

Closes #1034.

This PR builds on #885 and #1100, now merged into main, and adds checkpoint-aligned Kafka 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, topic identity, offsets, and retention-safe data-topic configuration, 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.
• The action-state topic uses cleanup.policy=compact,delete so explicit prefix deletion is available, with retention.ms=-1 and retention.bytes=-1 so Kafka cannot independently retire records required by a supported recovery point.

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. PR #1100 fixed typed action-state key identity. Recovery retains those key checks 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.
• Apply validates the action-state topic's cleanup and retention settings before COMMITTED and immediately before and after deletion.
• 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, unavailable target offsets, and unsafe data-topic retention settings 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.
• Plan files and control records reject trailing content after their JSON document.
• Control-topic replication factors outside 1–32767 are rejected before creating Kafka
  clients.

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
Production-created action-state topic settings, real Kafka control records, deleteRecords, beginning offsets, and boundary reads	KafkaActionStateCleanupCoordinatorIntegrationTest
Marker compatibility, boundary checks, Kafka identity, offset ranges, captured replay ends, and post-tombstone migration markers	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

Implementation details for traceability

• The action-state and control topics are distinct and dedicated to one job recovery history. New action-state topics use cleanup.policy=compact,delete, retention.ms=-1, and retention.bytes=-1; existing topics must be migrated to those settings before apply.
• 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.

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.

Jobs migrating from tombstones must first stop tombstone emission, restart from a recovery point valid under the existing tombstone trade-off, and complete a new checkpoint or savepoint. The first cleanup plan must use that post-tombstone recovery point so its marker follows every old tombstone.

Documentation

• doc-needed
• doc-not-needed
• 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?

• Yes
• No

Generated-by: OpenAI Codex CLI 0.153.0 (GPT-5)

[rob-9]

Rebased onto main after #885 and #1100 merged and cleaned up history. Fixed conflicts and ran a review pass.
The failing job looks unrelated to this PR's code changes. PTAL @wenjin272 @weiqingy @joeyutong , thanks!