[api][runtime][python] Add opt-in Kafka tombstones for pruned action state

api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java

@@ -81,6 +81,19 @@ public enum ConditionEvaluationFailureStrategy {
     public static final ConfigOption<Integer> KAFKA_ACTION_STATE_TOPIC_REPLICATION_FACTOR =
             new ConfigOption<>("kafkaActionStateTopicReplicationFactor", Integer.class, 1);
 
+    /**
+     * The config parameter determines whether pruning sends tombstone (null-valued) records to the
+     * Kafka action state topic so log compaction can reclaim pruned keys. Defaults to {@code
+     * false}: disabling this option does not invalidate older restore points through pruning, but
+     * the topic continues to grow. When enabled, the checkpoint whose completion triggers pruning
+     * remains usable, but restoring an earlier checkpoint or savepoint may replay tombstones
+     * written after that restore point, erasing action state the replay still needs and causing
+     * already completed actions to re-execute. Enable only if the job never restores from earlier
+     * checkpoints or savepoints, or if re-executing actions is acceptable.
+     */
+    public static final ConfigOption<Boolean> KAFKA_ACTION_STATE_TOMBSTONE_ENABLED =
+            new ConfigOption<>("kafkaActionStateTombstoneEnabled", Boolean.class, false);
+
     /** The config parameter specifies the Fluss bootstrap servers. */
     public static final ConfigOption<String> FLUSS_BOOTSTRAP_SERVERS =
             new ConfigOption<>("flussBootstrapServers", String.class, "localhost:9123");

docs/content/docs/operations/configuration.md

@@ -182,6 +182,7 @@ Here are the configuration options for Kafka-based Action State Store.
 | `kafkaActionStateTopic`             | (none)                   | String  | The config parameter specifies the Kafka topic for action state.            |
 | `kafkaActionStateTopicNumPartitions`| 64                       | Integer | The config parameter specifies the number of partitions for the Kafka action state topic. |
 | `kafkaActionStateTopicReplicationFactor` | 1                     | Integer | The config parameter specifies the replication factor for the Kafka action state topic. |
+| `kafkaActionStateTombstoneEnabled`  | false                    | Boolean | Whether pruning sends tombstone records so log compaction can reclaim pruned keys on a compacted action-state topic. Off by default: pruning does not invalidate older restore points, but the topic continues to grow. When enabled, the checkpoint whose completion triggers pruning remains usable, but restoring an earlier checkpoint or savepoint may replay later tombstones and re-execute already completed actions. Enable only if the job never restores from earlier checkpoints or savepoints, or if re-executing actions is acceptable. |
 
 #### Fluss-based Action State Store
 

docs/content/docs/operations/deployment.md

@@ -104,6 +104,10 @@ The same persisted action state is also used by fine-grained durable execution.
 
 See [Action State Store Configuration]({{< ref "docs/operations/configuration#action-state-store" >}}) for configuration options.
 
+{{< hint warning >}}
+**Note**: Enabling Kafka action-state tombstones can invalidate checkpoints or savepoints older than the prune and cause completed actions to execute again. See [Action State Store Configuration]({{< ref "docs/operations/configuration#action-state-store" >}}) for the recovery trade-off.
+{{< /hint >}}
+
 {{< hint info >}}
 **Note**: Exactly-once action consistency is guaranteed only if, after recovering from the same checkpoint, inputs for each key arrive in the same order as before recovery. If this ordering requirement is not met, the system falls back to exactly-once output consistency.
 {{< /hint >}}

python/flink_agents/api/core_options.py

@@ -140,6 +140,12 @@ class AgentConfigOptions:
         default=1,
     )
 
+    KAFKA_ACTION_STATE_TOMBSTONE_ENABLED = ConfigOption(
+        key="kafkaActionStateTombstoneEnabled",
+        config_type=bool,
+        default=False,
+    )
+
     FLUSS_BOOTSTRAP_SERVERS = ConfigOption(
         key="flussBootstrapServers",
         config_type=str,

runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java

@@ -78,6 +78,14 @@ void put(Object key, long seqNum, Action action, Event event, ActionState state)
     /**
      * Prune the state for a given key.
      *
+     * <p>Implementations must at least evict the matching entries from the in-memory cache. Whether
+     * the backend storage is also cleaned up is implementation-specific. Durable deletion can
+     * invalidate checkpoints or savepoints whose recovery markers precede the deletion: {@link
+     * #rebuildState(List)} replays the backend from the restored recovery marker, so records
+     * deleted after that marker may be state the replay still needs. Implementations must either
+     * enforce a recovery boundary that protects every supported restore point or clearly document
+     * the recovery trade-off of advancing beyond that boundary.
+     *
      * @param key the key whose state should be pruned
      * @param seqNum the sequence number up to which the state should be pruned
      */

runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java

@@ -52,6 +52,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.IntPredicate;
 
+import static org.apache.flink.agents.api.configuration.AgentConfigOptions.KAFKA_ACTION_STATE_TOMBSTONE_ENABLED;
 import static org.apache.flink.agents.api.configuration.AgentConfigOptions.KAFKA_ACTION_STATE_TOPIC;
 import static org.apache.flink.agents.api.configuration.AgentConfigOptions.KAFKA_ACTION_STATE_TOPIC_NUM_PARTITIONS;
 import static org.apache.flink.agents.api.configuration.AgentConfigOptions.KAFKA_ACTION_STATE_TOPIC_REPLICATION_FACTOR;
@@ -92,6 +93,9 @@ public class KafkaActionStateStore implements ActionStateStore {
     // Kafka topic that stores action states
     private final String topic;
 
+    // Whether pruning sends tombstone records for log compaction
+    private final boolean tombstoneEnabled;
+
     // When set, only records whose key-group is accepted by this predicate are kept in the
     // in-memory cache during rebuildState; null means retain all keys (default).
     private IntPredicate ownershipFilter;
@@ -113,6 +117,7 @@ public class KafkaActionStateStore implements ActionStateStore {
         this.topic = topic;
         this.latestKeySeqNum = new HashMap<>();
         this.agentConfiguration = agentConfiguration;
+        this.tombstoneEnabled = agentConfiguration.get(KAFKA_ACTION_STATE_TOMBSTONE_ENABLED);
         this.maxParallelism = maxParallelism;
     }
 
@@ -127,6 +132,7 @@ public KafkaActionStateStore(AgentConfiguration agentConfiguration, int maxParal
         this.actionStates = new HashMap<>();
         this.latestKeySeqNum = new HashMap<>();
         this.agentConfiguration = agentConfiguration;
+        this.tombstoneEnabled = agentConfiguration.get(KAFKA_ACTION_STATE_TOMBSTONE_ENABLED);
         this.topic =
                 Preconditions.checkNotNull(
                         agentConfiguration.get(KAFKA_ACTION_STATE_TOPIC),
@@ -256,17 +262,17 @@ public void rebuildState(List<Object> recoveryMarkers) {
                     break;
                 }
 
+                // Deserialization failures throw from poll() itself and are handled by the
+                // outer catch, so records here are always fully deserialized.
                 for (ConsumerRecord<String, ActionState> record : records) {
-                    try {
-                        if (!ActionStateUtil.isKeyRetained(ownershipFilter, record.key())) {
-                            continue;
-                        }
+                    if (!ActionStateUtil.isKeyRetained(ownershipFilter, record.key())) {
+                        continue;
+                    }
+                    if (record.value() == null) {
+                        // Tombstone record - remove the key from cache
+                        actionStates.remove(record.key());
+                    } else {
                         actionStates.put(record.key(), record.value());
-                    } catch (Exception e) {
-                        LOG.warn(
-                                "Failed to deserialize action state record: {}",
-                                record.value().toString(),
-                                e);
                     }
                 }
 
@@ -288,14 +294,50 @@ public void setOwnershipFilter(IntPredicate ownershipFilter) {
     public void pruneState(Object key, long seqNum) {
         LOG.debug("Pruning state for key: {} up to sequence number: {}", key, seqNum);
 
-        // Remove states from in-memory cache for this key up to the specified sequence
-        // number
-        actionStates
-                .keySet()
-                .removeIf(
-                        cachedKey ->
-                                ActionStateUtil.matchesBusinessKeyWithSeqNum(
-                                        cachedKey, key, stateSeqNum -> stateSeqNum <= seqNum));
+        // Collect state keys belonging to this key with sequence number <= seqNum.
+        List<String> keysToPrune = new ArrayList<>();
+        for (String stateKey : actionStates.keySet()) {
+            if (ActionStateUtil.matchesBusinessKeyWithSeqNum(
+                    stateKey, key, stateSeqNum -> stateSeqNum <= seqNum)) {
+                keysToPrune.add(stateKey);
+            }
+        }
+
+        // Send tombstones to Kafka so log compaction can reclaim storage; opt-in because
+        // tombstones break replay when restoring a checkpoint/savepoint older than the prune
+        // (see KAFKA_ACTION_STATE_TOMBSTONE_ENABLED). Send failures surface asynchronously,
+        // so report them via callback; the records then persist until manual cleanup.
+        if (tombstoneEnabled && producer != null && !keysToPrune.isEmpty()) {
+            try {
+                for (String stateKey : keysToPrune) {
+                    producer.send(
+                            new ProducerRecord<>(topic, stateKey, null),
+                            (metadata, exception) -> {
+                                if (exception != null) {
+                                    LOG.warn(
+                                            "Failed to send tombstone record for state key: {}. "
+                                                    + "The record will persist in the topic "
+                                                    + "until manual cleanup.",
+                                            stateKey,
+                                            exception);
+                                }
+                            });
+                }
+                LOG.debug(
+                        "Queued {} tombstone records to Kafka for key: {}",
+                        keysToPrune.size(),
+                        key);
+            } catch (Exception e) {
+                LOG.warn(
+                        "Failed to send tombstone records to Kafka for key: {}. "
+                                + "Records will persist in the topic until manual cleanup.",
+                        key,
+                        e);
+            }
+        }
+
+        // Remove from in-memory cache (always, regardless of tombstone success)
+        actionStates.keySet().removeAll(keysToPrune);
 
         LOG.debug("Pruned state for key: {} up to sequence number: {}", key, seqNum);
     }