rob-9 opened a new issue, #1099: URL: https://github.com/apache/flink-agents/issues/1099
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description ### 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: ```java 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 - Flink Agents: `0.4-SNAPSHOT` - Reproduced against `main` commit `19e22488` and PR #885 worktree commit `1f371bde`; both include the key format from PR #1024 - Apache Flink: `2.3.0` - Java: JDK 17 - OS: macOS 15.4.1, arm64 - No live Kafka or Fluss service is required; the Kafka pruning reproduction uses `MockProducer`, and the lookup reproduction exercises the stores' in-memory cache paths ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
