Search before asking
Description
ActionStateUtil.generateKey() persists the key-group computed from the original typed Flink key, but represents the logical business key using key.toString(). The matching helpers compare only that textual value.
Distinct Flink keys can therefore be treated as the same business key during lookup and pruning. Examples include Long(1) and String("1"), or distinct custom keys with the same toString() result. When the key-group also matches, their complete state keys can be identical.
Impact
- When distinct keys with the same string form are held by one store/subtask, Kafka and Fluss lookup cleanup can evict the other key's cached action state.
- Kafka pruning can emit tombstones for another key's exact records when both keys are held by the same store/subtask.
- Evicting recovered completed state can make the operator miss it on the next lookup, re-execute the action, and repeat external side effects.
- With
maxParallelism = 1, distinct keys with the same string form can produce identical complete state keys, allowing one key's state to overwrite or replay another key's state.
PR #1024 fixed recovery ownership by persisting the key-group computed from the typed key, but the business-key segment still stores only key.toString(), which does not preserve the key's type or uniquely identify distinct keys. When its option is enabled, PR #885 exposes the remaining problem more severely because tombstones make accidental deletion durable.
PR #1094 fixes a different part of the state key: the action UUID was derived from an Action.hashCode() that can change after a JVM restart. It does not change the key.toString() business-key identity or the matchesBusinessKey* helpers described here.
Proposed direction
Persist the key's serialized bytes, or a collision-resistant digest of them, and use that value consistently for key generation, lookup, pruning, and Kafka partitioning. Using the operator's Flink key serializer would make the durable identity follow the same typed representation used by keyed state. Comparing the key-group in addition to toString() would fix only cases where the keys belong to different groups; distinct keys can still share a key-group.
Define one explicit versioned format and do not interpret old records by matching key.toString(). If records written in the old format cannot be mapped to a Flink key unambiguously, recovery should fail with a clear compatibility error rather than guessing.
Acceptance criteria
Long(1) and String("1") remain isolated with maxParallelism = 128 and maxParallelism = 1.- Distinct custom keys with the same
toString() remain isolated.
- Key identity remains stable across serialization and task recovery.
- Kafka and Fluss lookup, divergence cleanup, and pruning use the same serialized key identity.
- Pruning one key never emits a tombstone for another key.
- Regression tests cover Kafka, Fluss, and the operator recovery path.
How to reproduce
Using the existing NoOpAction and InputEvent test fixtures:
Action action = new NoOpAction("test-action");
InputEvent event = new InputEvent("test-input");
String numericKey = ActionStateUtil.generateKey(1L, 1L, action, event, 128);
String stringKey = ActionStateUtil.generateKey("1", 1L, action, event, 128);
assertNotEquals(numericKey, stringKey);
assertTrue(ActionStateUtil.matchesBusinessKey(stringKey, 1L));The keys had different key-groups (Long(1) → 86, String("1") → 54), but the cross-key match passed.
With both records cached, KafkaActionStateStore.pruneState(1L, 1L) evicted both and emitted two tombstones. A Long(1) cache-miss lookup also evicted a newer String("1") record in both Kafka and Fluss.
With maxParallelism = 1, the two generated state keys were identical.
Expected: distinct Flink keys remain isolated regardless of their string representation.
Actual: keys with equal toString() values cross-match; when their key-group also matches, their complete state keys collide.
These helper and store paths were executed directly. The resulting action re-execution follows the operator's cache-miss path but was not reproduced end to end through a Flink restart.
Version and environment
Are you willing to submit a PR?
Search before asking
Description
ActionStateUtil.generateKey()persists the key-group computed from the original typed Flink key, but represents the logical business key usingkey.toString(). The matching helpers compare only that textual value.Distinct Flink keys can therefore be treated as the same business key during lookup and pruning. Examples include
Long(1)andString("1"), or distinct custom keys with the sametoString()result. When the key-group also matches, their complete state keys can be identical.Impact
maxParallelism = 1, distinct keys with the same string form can produce identical complete state keys, allowing one key's state to overwrite or replay another key's state.PR #1024 fixed recovery ownership by persisting the key-group computed from the typed key, but the business-key segment still stores only
key.toString(), which does not preserve the key's type or uniquely identify distinct keys. When its option is enabled, PR #885 exposes the remaining problem more severely because tombstones make accidental deletion durable.PR #1094 fixes a different part of the state key: the action UUID was derived from an
Action.hashCode()that can change after a JVM restart. It does not change thekey.toString()business-key identity or thematchesBusinessKey*helpers described here.Proposed direction
Persist the key's serialized bytes, or a collision-resistant digest of them, and use that value consistently for key generation, lookup, pruning, and Kafka partitioning. Using the operator's Flink key serializer would make the durable identity follow the same typed representation used by keyed state. Comparing the key-group in addition to
toString()would fix only cases where the keys belong to different groups; distinct keys can still share a key-group.Define one explicit versioned format and do not interpret old records by matching
key.toString(). If records written in the old format cannot be mapped to a Flink key unambiguously, recovery should fail with a clear compatibility error rather than guessing.Acceptance criteria
Long(1)andString("1")remain isolated withmaxParallelism = 128andmaxParallelism = 1.toString()remain isolated.How to reproduce
Using the existing
NoOpActionandInputEventtest fixtures:The keys had different key-groups (
Long(1)→ 86,String("1")→ 54), but the cross-key match passed.With both records cached,
KafkaActionStateStore.pruneState(1L, 1L)evicted both and emitted two tombstones. ALong(1)cache-miss lookup also evicted a newerString("1")record in both Kafka and Fluss.With
maxParallelism = 1, the two generated state keys were identical.Expected: distinct Flink keys remain isolated regardless of their string representation.
Actual: keys with equal
toString()values cross-match; when their key-group also matches, their complete state keys collide.These helper and store paths were executed directly. The resulting action re-execution follows the operator's cache-miss path but was not reproduced end to end through a Flink restart.
Version and environment
0.4-SNAPSHOTmaincommit19e22488and PR [api][runtime][python] Add opt-in Kafka tombstones for pruned action state #885 worktree commit1f371bde; both include the key format from PR [Bug] Recovered Durable ActionState for keys owned by other subtasks is never pruned #10242.3.0MockProducer, and the lookup reproduction exercises the stores' in-memory cache pathsAre you willing to submit a PR?