[runtime][api][python] Add checkpoint-aligned Kafka action-state cleanup by rob-9 · Pull Request #1101 · apache/flink-agents

rob-9 · 2026-09-04T08:45:06Z

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

Each checkpoint stores a versioned marker containing the physical Kafka topic ID and next read offset for every action-state partition.
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.
After reviewing the topic identity, partitions, and offsets, the operator runs apply.
Apply validates the plan and current Kafka state, writes a COMMITTED record to a dedicated compacted control topic, and calls deleteRecords.
Apply verifies the physical topic identity and resulting beginning offsets, then writes APPLIED. A COMMITTED operation can safely resume after interruption.
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.

Implementation details for traceability

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.

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
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

- add kafkaActionStateTombstoneEnabled (default false) so tombstone emission is opt-in; rebuildState still honors tombstones already in the topic regardless of the flag - match the parsed key part exactly in pruneState so pruning key "a_1" can no longer tombstone state of the distinct key "a" - report async tombstone send failures via producer callback (flush() does not surface per-record errors) - narrow the prune catch to IllegalArgumentException and state the retention consequence in the warning - remove dead inner try/catch in rebuildState (deserialization errors throw from poll(), not from the map ops it wrapped) - document the durable-deletion replay constraint on ActionStateStore.pruneState and add the new option to the config docs - tests: default-off pruning, prefix-collision regression, tombstone replay in rebuildState; simplify assertions

…neState - testPruneStateEvictsCacheEvenWhenTombstoneSendFails: verifies pruneState degrades gracefully and still evicts the in-memory entry when a tombstone send fails asynchronously (the callback-reporting fix from the prior commit) - testPruneStateSkipsUnparseableKeys: verifies a state key that cannot be parsed into 4 parts is retained rather than pruned (the narrowed IllegalArgumentException catch) Both were verified to fail when the corresponding fix is reverted.

…neState to cache eviction only

rob-9 · 2026-09-04T20:46:07Z

the job failure here looks unrelated to the PR.

rob-9 and others added 24 commits July 7, 2026 17:22

fix: handle null-valued records in KafkaActionStateStore.rebuildState()

0db12a6

feat: send tombstone records in KafkaActionStateStore.pruneState()

98b1aaa

style: fix spotless formatting violations

a02a28a

Merge fork/main into feat/691-kafka-prune-tombstones

55f7874

feat: add kafkaActionStateTombstoneEnabled to python core options

41c9fee

test: exercise the real async tombstone send failure path

07958ff

docs: document that keys containing '_' are never pruned or tombstoned

f46cffc

refactor: drop redundant producer.flush() in pruneState

2cb557c

docs: clarify tombstone recovery scope

2cb89d8

test: keep tombstone assertions in the dedicated test, revert testPru…

14b894f

…neState to cache eviction only

[runtime][docs] Harden tombstone recovery edge cases

16b44b6

[runtime][java] Protect Fluss cleanup from key collisions

d09d94c

[runtime][java] Strengthen unparseable Kafka key test

a8bf18a

[runtime][java] Quiet repeated Kafka key parse failures

fca59c2

[api][runtime][docs] Clarify action state key limitation

8b8db63

[docs] Restore exactly-once action wording

a6c68e3

[runtime][java] Scope Kafka divergence cleanup by key

298e44b

[runtime][java] Quiet retained Fluss state keys

53ca826

[docs] Restore action recovery explanation

3435e07

[runtime][java] Scope divergence detection by key

ad61517

[runtime][java] Support underscores in action state keys

69fa856

[runtime] Merge main into Kafka tombstone branch

1f371bd

github-actions Bot added doc-included Your PR already contains the necessary documentation updates. fixVersion/0.4.0 priority/major Default priority of the PR or issue. labels Sep 4, 2026

rob-9 added 3 commits September 4, 2026 11:35

[api][java][python] Add Kafka cleanup boundary configuration

92cb383

[runtime][java] Add checkpoint-aligned Kafka cleanup

a14b420

[docs] Document checkpoint-aligned Kafka cleanup

9aed69c

rob-9 added 3 commits September 4, 2026 11:40

[runtime][java] Harden Kafka cleanup safety checks

748dc22

[docs] Clarify Kafka cleanup CLI workflow

155567d

[test][java] Fix Kafka Testcontainers classpath

d075edd

rob-9 force-pushed the feat/checkpoint-aligned-kafka-cleanup branch from e71972e to d075edd Compare September 4, 2026 19:10

Navigation Menu

Uh oh!

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

[runtime][api][python] Add checkpoint-aligned Kafka action-state cleanup#1101
rob-9 wants to merge 30 commits into
apache:mainfrom
rob-9:feat/checkpoint-aligned-kafka-cleanup

rob-9 commented Sep 4, 2026

Uh oh!

rob-9 commented Sep 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

rob-9 commented Sep 4, 2026

Purpose of change

Runtime flow

Key decisions

Related work

Behavioral Semantics

Interaction decisions

Behavioral contracts

Failure behavior

Tests

API

Documentation

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

rob-9 commented Sep 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant