[Feature] Kafka ActionStateStore should clean up pruned ActionState records
- Dominant language
- Java
- Stars
- 452
- Forks
- 167
- Avg merge
- 5d 9h
- Merged PRs (30d)
- 49
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar.
### Description
### Problem
ActionState is persisted to the external ActionStateStore to support durable execution recovery. The current production store implementation is Kafka.
When a checkpoint completes
```java
actionStateStore.pruneState(entry.getKey(), entry.getValue());
```
The intent is to prune action states for a Flink key up to the completed sequence number. However, the current KafkaActionStateStore.pruneState(...) implementation only removes matching entries from the in-memory actionStates cache. It does not delete the corresponding records from the Kafka topic. These historical keys are no longer useful for recovery once the completed sequence number has been covered by a successful checkpoint, but they can still accumulate indefinitely in Kafka.
### Solution
After a checkpoint has completed,should delete from kafkaActionStateStore. Because the Kafka topic is configured with cleanup.policy=compact, writing a tombstone record (key = stateKey, value = null) allows Kafka log compaction to eventually remove all historical records for that ActionState key.
```java
producer.send(new ProducerRecord<>(topic, stateKey, null));
```
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
Research direction
Start at KafkaActionStateStore.pruneState(...) and trace how stateKey and the Kafka producer are handled after a checkpoint completes. Confirm the topic uses cleanup.policy=compact, then ensure pruned records are represented by tombstones with the state key and a null value. Done means historical ActionState records can be compacted from Kafka while the in-memory cache is still pruned.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- distributed-systems
- Issue type
- Feature
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100