weiqingy commented on code in PR #885:
URL: https://github.com/apache/flink-agents/pull/885#discussion_r3601149105
##########
api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java:
##########
@@ -64,6 +64,19 @@ public class AgentConfigOptions {
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}: without tombstones the topic grows unboundedly, but restoring
any checkpoint or
+ * savepoint replays correctly. When enabled, restoring from the latest
completed checkpoint is
+ * unaffected, but restoring an older 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 non-latest
+ * 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);
Review Comment:
Defaulting to `false` reads as the right call to me, and I think for a
sharper reason than the javadoc gives itself credit for: with
`cleanup.policy=compact` on the topic (`KafkaActionStateStore.java:396`), a
tombstone is a durable delete instruction to the log cleaner — so the
older-checkpoint erasure you documented above isn't something the replay side
could have absorbed. Gating emission is the lever that actually controls it. A
safe default-on would seem to need emission gated on the oldest
still-restorable checkpoint, and I can't see a cheap way to know that here, so
I'm not suggesting you flip it.
That does leave #691's original ask — the unbounded growth — unaddressed
while the option is off. Where do you see this landing: is opt-in the endpoint,
or would a follow-up be worth filing so the issue has somewhere to point?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -276,30 +282,64 @@ public void rebuildState(List<Object> recoveryMarkers) {
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
- .entrySet()
- .removeIf(
- entry -> {
- String stateKey = entry.getKey();
- // Extract key and sequence number from the state
key
- // State key format: "key_seqNum_action_event"
- if (stateKey.startsWith(key.toString() + "_")) {
- try {
- List<String> parts =
ActionStateUtil.parseKey(stateKey);
- if (parts.size() >= 2) {
- long stateSeqNum =
Long.parseLong(parts.get(1));
- return stateSeqNum <= seqNum;
- }
- } catch (NumberFormatException e) {
+ // Collect state keys belonging to this key with sequence number <=
seqNum. The parsed
+ // key part must match exactly: prefix matching alone would let
pruning key "a_1" match
+ // state keys of the distinct key "a" (whose keys also start with
"a_1_").
+ String keyStr = key.toString();
+ String keyPrefix = keyStr + "_";
+ List<String> keysToPrune = new ArrayList<>();
+ for (String stateKey : actionStates.keySet()) {
+ if (!stateKey.startsWith(keyPrefix)) {
+ continue;
+ }
+ try {
+ List<String> parts = ActionStateUtil.parseKey(stateKey);
+ if (parts.get(0).equals(keyStr) &&
Long.parseLong(parts.get(1)) <= seqNum) {
+ keysToPrune.add(stateKey);
+ }
+ } catch (IllegalArgumentException e) {
+ LOG.warn(
+ "Cannot parse state key: {}. The entry cannot be
pruned and will be "
+ + "retained in memory and in the topic.",
+ stateKey,
+ e);
+ }
+ }
+
+ // 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 parse sequence number
from state key: {}",
- stateKey);
+ "Failed to send tombstone record
for state key: {}. "
+ + "The record will persist
in the topic "
+ + "until manual cleanup.",
+ stateKey,
+ exception);
}
- }
- return false;
- });
+ });
+ }
+ producer.flush();
Review Comment:
Does this flush earn its place? The callback just above already reports
failures, the eviction below runs regardless of the outcome, and ordering looks
covered — `put()` flushes before returning (line 146), so a tombstone is always
sent after its record is acked.
Curious whether there's a case it's protecting that I'm not seeing.
##########
docs/content/docs/operations/configuration.md:
##########
@@ -168,6 +168,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. Off by default: without tombstones the topic grows unboundedly, but
restoring any checkpoint or savepoint replays correctly. When enabled,
restoring from the latest completed checkpoint is unaffected, but restoring an
older checkpoint or savepoint may replay tombstones written after that restore
point and re-execute already completed actions. Enable only if the job never
restores from non-latest checkpoints or savepoints, or if re-executing actions
is acceptable. |
Review Comment:
Small one: this row and the new public
`AgentConfigOptions.KAFKA_ACTION_STATE_TOMBSTONE_ENABLED` are both in the diff,
while the description checks `doc-not-needed` and says "No public API changes".
It also still describes the unconditional design rather than the opt-in one
that shipped. Mind refreshing it before merge?
##########
runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java:
##########
@@ -196,6 +207,129 @@ void testPruneState() throws Exception {
assertNull(
actionStates.get(ActionStateUtil.generateKey(TEST_KEY, 2L,
testAction, testEvent)));
assertNotNull(actionStateStore.get(TEST_KEY, 3L, testAction,
testEvent));
+
+ // Assert - tombstones should have been sent to Kafka
+ var history = mockProducer.history();
+ assertThat(history).hasSize(2);
Review Comment:
`testPruneStateSendsTombstonesWithCorrectKeys` pins this same contract more
precisely — `containsExactlyInAnyOrder(key1, key2)` on the keys,
`containsOnlyNulls()` on the values. A bug in tombstone keys, count, or
nullness would fail both together, so I'm not sure this block catches anything
the dedicated test would miss.
Would it read cleaner to leave `testPruneState` as it was — default store,
cache eviction only, which also drops the `tombstoneEnabledStore` swap at the
top — and keep the tombstone contract in one place?
`testPruneStateNoTombstonesByDefault` already pins the default-off behavior.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -276,30 +282,64 @@ public void rebuildState(List<Object> recoveryMarkers) {
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
- .entrySet()
- .removeIf(
- entry -> {
- String stateKey = entry.getKey();
- // Extract key and sequence number from the state
key
- // State key format: "key_seqNum_action_event"
- if (stateKey.startsWith(key.toString() + "_")) {
- try {
- List<String> parts =
ActionStateUtil.parseKey(stateKey);
- if (parts.size() >= 2) {
- long stateSeqNum =
Long.parseLong(parts.get(1));
- return stateSeqNum <= seqNum;
- }
- } catch (NumberFormatException e) {
+ // Collect state keys belonging to this key with sequence number <=
seqNum. The parsed
+ // key part must match exactly: prefix matching alone would let
pruning key "a_1" match
+ // state keys of the distinct key "a" (whose keys also start with
"a_1_").
+ String keyStr = key.toString();
+ String keyPrefix = keyStr + "_";
+ List<String> keysToPrune = new ArrayList<>();
+ for (String stateKey : actionStates.keySet()) {
+ if (!stateKey.startsWith(keyPrefix)) {
+ continue;
+ }
+ try {
+ List<String> parts = ActionStateUtil.parseKey(stateKey);
+ if (parts.get(0).equals(keyStr) &&
Long.parseLong(parts.get(1)) <= seqNum) {
Review Comment:
The collision you caught here feels like a thread worth pulling one turn
further — I think the same root cause has a second face that this check can't
reach.
Both this guard and the collision it fixes trace back to `generateKey`
joining on `_` while embedding the raw key unescaped
(`ActionStateUtil.java:44-56`). That's what let pruning key `a_1` match agent
key `a`'s state key `a_1_<uuid>_<uuid>`, and the equality check handles that
case correctly.
The sibling case is when the *agent key itself* contains `_`. For an
ordinary `keyBy("user_123")` the state key is `user_123_1_<uuid>_<uuid>`, so
`split("_")` yields 5 parts (UUIDs use hyphens) and `parseKey`'s
`checkArgument(parts.length == 4)` throws (`ActionStateUtil.java:58-63`). This
line never gets evaluated, and the entry lands in the catch on line 300 — so
it's never evicted from the cache, never tombstoned, and never reclaimed by
compaction. For any key containing `_`, the growth this PR targets would stay
as it is. `FlussActionStateStore.removeStateEntries` catches `Exception` and
skips the same way (`FlussActionStateStore.java:242-261`).
Worth saying that catching `IllegalArgumentException` on line 300 is an
improvement in its own right — the old code caught only
`NumberFormatException`, so a bad part count escaped `pruneState` and could
take down `notifyCheckpointComplete`. Warn-and-retain is strictly better.
Escaping the separator properly would touch three files, which feels like
more than this PR signed up for — would you rather note the limitation in the
option's javadoc and file a follow-up? Or do you see a narrower angle I'm
missing?
And on the test: `testPruneStateSkipsUnparseableKeys` pins this branch with
`TEST_KEY + "_1_onlythreeparts"`. Any appetite for a realistic fixture like
`"user_123"` — same branch, but it'd document the case that actually reaches it?
--
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]
