CodeTrainerMan commented on issue #1084:
URL: https://github.com/apache/flink-agents/issues/1084#issuecomment-5767599098
Thanks for the design work here. I read the latest proposal against
`ActionStateUtil` on `main` and found three things that I think need settling
before any code lands. I am not claiming this issue -- just hoping to save the
implementer a round trip.
**1. `generateKey()` has to stay a pure function; a self-incrementing
counter inside it breaks that.**
The proposal tracks "the number of times `generateKey()` is called for this
tuple" and appends it. But `generateKey()` is called from *both* directions of
the store, not only when a new state is written:
- `KafkaActionStateStore.put()` (L160) and `.get()` (L178)
- `FlussActionStateStore.put()` (L215) and `.get()` (L238)
`ActionExecutionOperator.processActionTaskForKey()` calls
`durableExecManager.maybeGetActionState(...)` (L513-515) *before* deciding
whether to execute, and only writes afterwards. If the ordinal is bumped inside
`generateKey()`, the lookup that misses consumes ordinal N and the following
write gets N+1, so the state lands under a key that no subsequent `get()` can
derive. Durable execution would silently stop hitting its own cache, and the
symptom is "state never found" rather than an error.
The ordinal therefore needs to be passed *into* `generateKey()` by the
caller (or folded into the event identity), keeping the method deterministic
and idempotent the way it is today.
**2. A sixth segment makes every already-persisted action state
unrecoverable.**
The layout is fixed at `KEY_SEGMENT_COUNT = 5`, and `isKeyRetained()`
hard-fails rather than skips on any key it cannot attribute:
```java
String[] parts = splitValidatedKey(stateKey);
if (parts == null) {
throw new IllegalStateException(
"Malformed action-state key during recovery: expected five
fields. Key: " + ...);
}
```
`isKeyRetained()` runs over every record during recovery --
`KafkaActionStateStore.poll()` (L277) and `FlussActionStateStore` (L455). That
throwing behavior is pinned by
`ActionStateUtilTest.testIsKeyRetainedRejectsUnrecognizedFormatKeys`, which
asserts that even a bare `"malformed-key"` throws. This differs from
`matchesBusinessKeyIdentity*`, which return `false` under the documented "never
prune what cannot be attributed" rule; `isKeyRetained()` has no such fallback.
So a five-to-six segment change turns every pre-existing Kafka/Fluss action
state, and anything in an older savepoint, into a hard recovery failure on
upgrade -- the opposite of what durable execution exists for. Since Kafka/Fluss
are external stores, there is no serializer-snapshot upgrade path to lean on
either.
A format-preserving option: keep five segments and fold the occurrence
ordinal into the event segment's derivation, e.g.
`UUID.nameUUIDFromBytes(attributesBytes + ordinal)`, with the ordinal supplied
by the caller. Old keys stay parseable and simply behave as the first
occurrence.
**3. The counter re-introduces the collision on recovery -- @wenjin272's
point (2) in another form.**
An ordinal derived from call order is still an order-dependent resource. On
recovery, a sibling whose state is already completed takes the skip-and-replay
branch at `ActionExecutionOperator` L518 (`actionState != null &&
actionState.isCompleted()`): it only calls `get()`, never `put()`. A sibling
that had *not* completed must re-execute and `put()`. The sequence in which
ordinals are consumed therefore differs from the original run, and a
re-executed sibling can land on the ordinal that belonged to its completed
sibling -- look up that sibling's completed state and replay its output. That
is precisely the bug this issue is trying to fix, arriving through the new
mechanism.
This is why the lineage-based identity sketched on Sep 14 (which @wenjin272
said looked reasonable) seems the safer direction: identity derived from
*where* an event sits in the causal structure, carried with the pending event
and checkpointed with it, rather than from *when* a call happens.
Happy to be corrected if I have misread the recovery path.
--
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]
