Ashfaqbs commented on issue #1084: URL: https://github.com/apache/flink-agents/issues/1084#issuecomment-5548531247
Traced the root cause through the actual dispatch code to pin down exactly when this collision can occur, in case it's useful for whoever picks this up: `ActionStateUtil.generateUUIDForEvent()` hashes only `event.getAttributes()`. Looking at `ActionExecutionOperator`, `stateManager.initOrIncSequenceNumber()` (line ~311) is only called when `isInputEvent` is true — i.e. once per top-level `InputEvent` for a key. Every action task triggered as part of the resulting cascade (including actions triggered by internally-generated events, not just the original input) shares that same `(key, seqNum)` until the next top-level `InputEvent` arrives. So the collision isn't limited to "two top-level events with identical attributes" — it also happens when a single action emits two *internal* events with identical attributes (e.g. differing only in an attachment payload, which isn't part of `getAttributes()`) that later reach the same downstream action within the same run. This also explains why `Event.getId()` alone can't be the fix (as the issue already notes): it's not just about disambiguating two independent event occurrences, it's specifically about disambiguating siblings generated *within the same seqNum round*, in a way that's stable across replay. **Proposed design**, for discussion before I write any code: - Add a new keyed `ValueState<Long>` (same shape as the existing `sequenceNumberKState` in `OperatorStateManager`) that resets to `0` each time `initOrIncSequenceNumber()` fires, and increments by one each time an action task is *enqueued* for processing within that round. - Fold that ordinal into the composite state key (`ActionStateUtil.generateKey`) alongside the existing `seqNum` segment, so two sibling events with identical attributes get distinct keys even when generated in the same round. - Backward compatibility isn't a concern per the existing code comment in `ActionStateUtil` — pre-format durable state is already not preserved across key-format changes. **Open question I'd like a maintainer's read on before implementing:** this only works if action-task enqueue order is guaranteed deterministic across replay for a given key (so the ordinal lines up the same way on recovery as it did on the original run). From reading the single-threaded per-key processing queue, that looks true today, but I'd rather confirm that invariant is intentional and expected to hold going forward before building a durable-state key format around it. Happy to pick this up once there's agreement on the approach — the repro test in the issue description already gives a solid starting point for the regression test. -- 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]
