weiqingy commented on code in PR #885:
URL: https://github.com/apache/flink-agents/pull/885#discussion_r3837521712
##########
runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java:
##########
@@ -121,6 +121,17 @@ void testGetTriggersDivergenceCleanup() throws Exception {
assertThat(store.get(TEST_KEY, 3L, testAction, testEvent)).isNull();
}
+ @Test
+ void testPruneStateRetainsKeyContainingUnderscore() throws Exception {
+ String agentKey = "user_123";
+ String stateKey = ActionStateUtil.generateKey(agentKey, 1L,
testAction, testEvent);
+ actionStates.put(stateKey, testActionState);
+
+ store.pruneState(agentKey, 1L);
+
+ assertThat(actionStates).containsKey(stateKey);
Review Comment:
This pins one half of the Kafka fix on the Fluss side. The other half, the
exact first-part check at `KafkaActionStateStore.java:301`, doesn't look like
it came across.
`FlussActionStateStore.pruneState:489` calls
`removeStateEntries(key.toString() + "_", stateSeqNum -> stateSeqNum <=
seqNum)`, and `removeStateEntries:243` filters on `startsWith(keyPrefix)` plus
the parsed sequence number, with no comparison of `parts.get(0)` against the
key being pruned.
So Flink key `a` at seq 1 stores `a_1_<eventUuid>_<actionUuid>`, which
parses cleanly as 4 parts with `parts.get(1) == "1"`. Pruning the distinct key
`a_1` at any seqNum >= 1 matches the prefix `a_1_` and satisfies `1 <= seqNum`,
so key `a`'s completed state is evicted. `FlussActionStateStore.get:227` reads
the cache only, so the next lookup returns null and the action re-runs.
Is leaving Fluss on the prefix match deliberate, or would the same
`parts.get(0)` check belong in `removeStateEntries`?
##########
runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java:
##########
@@ -142,6 +165,16 @@ void testGetActionStateWithDiverge() throws Exception {
assertNull(actionStateStore.get(TEST_KEY, 4L, testAction, testEvent));
}
+ @Test
+ void testGetRetainsUnparseableKey() throws Exception {
+ String flinkKey = "user_123";
+ String stateKey = ActionStateUtil.generateKey(flinkKey, 1L,
testAction, testEvent);
+ actionStates.put(stateKey, testActionState);
+
+ assertThat(actionStateStore.get(flinkKey, 2L, testAction,
testEvent)).isNull();
Review Comment:
Does this discriminate the way its name suggests? The entry is stored at seq
1 and looked up at seq 2. The divergence cleanup only evicts entries with
`stateSeqNum > seqNum` (`KafkaActionStateStore.java:182`), and 1 is not greater
than 2, so the entry is retained whether or not `parseKey` throws. The lookup
returns null either way, since no seq-2 key was ever stored, so both assertions
hold against an implementation that parses `user_123` correctly.
Inverting the two sequence numbers would separate the cases, in case that's
useful: store at seq 3, then `get(flinkKey, 1L, ...)`. Correct parsing evicts
(3 > 1), only the parse failure retains it, and the seq-1 lookup is still a
cache miss so the `removeIf` fires.
`testPruneStateSkipsUnparseableKeys` and the Fluss twin both do fail against
a fixed implementation, so this looks like the odd one out.
##########
docs/content/docs/operations/deployment.md:
##########
@@ -92,7 +92,7 @@ After recovery from a checkpoint, Flink Agents reprocess
events that arrived aft
### Exactly-Once Action Consistency
-To ensure exactly-once action consistency, you must configure an external
action state store. Flink Agents record action state in this store on a
per-action basis. After recovering from a checkpoint, Flink Agents consult the
external store and will not re-execute actions that were already completed.
This guarantees each action is executed exactly once after recovering from a
checkpoint.
+To ensure exactly-once action consistency, you must configure an external
action state store. Flink Agents record action state in this store on a
per-action basis. After recovering from a checkpoint, Flink Agents consult the
external store and reuse completed action state when its backing record remains
available. This prevents re-execution for checkpoints supported by the store's
retained recovery history.
Review Comment:
nit: this paragraph covers both backends, including Fluss and the default
Kafka setup with tombstones off. The hazard it now hedges for is opt-in and
Kafka-only, and the `hint warning` box a few lines below states it with that
scope. "backing record remains available" and "the store's retained recovery
history" also appear nowhere else in either doc, so a reader has nothing to
resolve them against.
Is the hedge doing work here that the warning box below doesn't already do?
One way this could read is with the original sentence restored: "This
guarantees each action is executed exactly once after recovering from a
checkpoint."
##########
docs/content/docs/operations/configuration.md:
##########
@@ -170,6 +170,8 @@ The eight `memory.generate-event*` options have no raw
`ConfigOption` default. W
|------------------------------|------------------|---------|------------------------------------------------------------------------------------------|
| `actionStateStoreBackend` | (none) | String | The backend for
action state store. Supported values: `"kafka"`, `"fluss"`. |
+Durable action state stores currently join raw Flink keys and other key parts
with an unescaped `_`. Flink keys containing `_` cannot be parsed safely during
pruning, so both Kafka and Fluss retain their state in memory and backend
storage. Kafka also emits no tombstones for those keys.
Review Comment:
nit: "both Kafka and Fluss retain their state in memory and backend storage"
reads as though backend retention follows from the parse failure. On the Fluss
side it doesn't. `FlussActionStateStore.pruneState:487` never deletes from the
backend for any key, parseable or not, and its javadoc says why: "The Fluss log
is append-only; physical cleanup relies on Fluss log retention configuration."
Two smaller things alongside it. All three doc sites frame the limitation as
growth, but the same parse failure also makes the divergence cleanup in `get()`
inert (`KafkaActionStateStore.java:184`, `FlussActionStateStore.java:257`), so
stale higher-seq state survives a divergence that was actually detected. That's
a correctness cost rather than a storage one.
And #1034 covers the checkpoint boundary, but nothing seems to track the key
encoding itself. Worth its own issue?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -175,10 +181,13 @@ public ActionState get(Object key, long seqNum, Action
action, Event event) thro
// the requested seqNum
return stateSeqNum > seqNum;
}
- } catch (NumberFormatException e) {
+ } catch (IllegalArgumentException e) {
LOG.warn(
- "Failed to parse sequence number
from state key: {}",
- stateKey);
+ "Cannot parse state key: {}. The
entry cannot be "
+ + "considered for
divergence cleanup and will "
+ + "be retained.",
+ entry.getKey(),
Review Comment:
The widening isn't what I'm asking about here, it's how often this line can
fire.
The `removeIf` scans the whole cache whenever
`!actionStates.containsKey(stateKey) || hasDivergence` (`:170`), `get()` sits
on the durable-execution path with three call sites in
`DurableExecutionManager` (`:212`, `:218`, `:255`), and entries that fail to
parse are retained by design. So one `user_123` entry produces a WARN on every
subsequent cache miss for the life of the job, each with a full stack trace,
since `e` is passed as a trailing arg.
`pruneState:305` already warns for the same key on the prune path. Would
dropping the throwable here, or demoting this to DEBUG, cost signal you'd want
to keep?
##########
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:
Confirmed, closed on my side.
##########
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:
Checked the updated body and docs, this one's closed.
##########
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:
Closed on my side, thanks.
##########
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:
Saw #1034, that answers this. It's the right home for what #691 still has
open, so I'll follow the discussion there.
--
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]
