Ashfaqbs opened a new pull request, #1342: URL: https://github.com/apache/flink-agents/pull/1342
Linked issue: #1084 ### Purpose of change Two distinct event occurrences reaching the same action under the same business key and sequence number, with identical attributes, previously collided on the same durable action-state key. The second occurrence's pre-execution lookup could then find the first occurrence's just-persisted completed state and incorrectly skip its own execution, replaying the first occurrence's output instead. This fixes that: sibling occurrences with identical attributes now always get distinct keys, while a changed attribute still changes the key exactly as before (content-based replay-divergence detection is unchanged). #### Runtime flow 1. Root event ingestion (`ActionExecutionOperator#processEvent`, `isInputEvent` branch): after the per-key sequence number is incremented, `assignRootOccurrenceId` derives `occurrenceId = sha256(businessKeyIdentity + "#" + seqNum)` and assigns it on the event. 2. Output emission (`processActionTask` on the serial engine; `Work#commit` on the parallel engine): `assignOutputOccurrenceIds` computes the producing action execution's own composite key (via the triggering event's already-assigned `occurrenceId`), then assigns each output event `occurrenceId = sha256(producerExecutionId + "#" + ordinal)`, `ordinal` being its 0-based position among that execution's emitted outputs — before any output is dispatched or persisted. 3. `ActionStateUtil#generateUUIDForEvent` consumes `event.getOccurrenceId()`: when present, it is folded together with (not in place of) the existing attributes hash into the UUID segment; when absent (state persisted by a pre-fix version), it falls back unchanged to the attributes-only hash. #### Key decisions - **Carrier**: a new non-contract `Event` field (`occurrenceId`), mirroring the existing `sourceTimestamp` field, rather than a wrapper type. This avoids touching every `ActionTask`/`SubgraphEvent`/serializer call site a wrapper would require, and survives Flink's generic (Kryo-based) `Event` serializer across checkpoints the same way `sourceTimestamp` already does. - **Fold with content, don't replace it**: an occurrence identity that ignored attributes entirely (rejected) regressed `ActionStateDivergenceRecoveryTest`, letting a later restore with a changed input reuse the other input's cached result at the same sequence number. Folding lineage together with the attributes hash keeps that protection while adding sibling disambiguation. - **Reuse the store's own key encoder**: identity derivation goes through two new `ActionStateStore` methods (`generateBusinessKeyIdentity`, `generateActionExecutionId`) that delegate to the same `ActionStateKeyEncoder` each store already holds, rather than duplicating key-derivation logic in the operator — the composite key format and the occurrence identity can never drift apart. - No-op when durable execution is disabled, since nothing keys state on it in that case. ### Behavioral Semantics #### Interaction decisions | Durable execution | Event kind | `occurrenceId` | Resulting key segment | |---|---|---|---| | disabled | any | not assigned | unchanged: attributes hash only | | enabled | root (new) | `businessKey#seqNum` | folded with attributes | | enabled | output (new) | `producerExecId#ordinal` | folded with attributes | | enabled | any, pre-fix persisted state | absent (`null`) | unchanged: attributes hash only | #### Behavioral contracts 1. Two sibling output events emitted by the same action execution with identical attributes get distinct action-state keys. 2. A replayed root `InputEvent` (same business key, same sequence number, fresh random `Event#getId()`) derives the same occurrence identity, and therefore the same key, as the original. 3. A changed attribute value still changes the key even when the lineage position (producer + ordinal, or business key + seqNum) is unchanged. 4. The composite key format (five segments, 36-character UUID per identity segment) is unchanged; `parseKey`/`isKeyRetained`/`businessKeyIdentityOf` behavior is unchanged. 5. An event never assigned an occurrence identity (state persisted by an older version) keys exactly as before the fix. #### Failure behavior `generateBusinessKeyIdentity`/`generateActionExecutionId` return `null` when no durable store is configured; callers treat `null` as "skip assignment" and never throw. `deriveOutputEventOccurrenceId` rejects a negative ordinal (`IllegalArgumentException`) and a null producer id (`NullPointerException`) defensively, though the operator's own call site cannot reach either. No new exception types; `IOException` propagation through `generateKey` is unchanged. ### Tests | Contract | Test | |---|---| | 1. Sibling collision fixed | `ActionStateUtilTest#testDistinctEventOccurrencesWithOccurrenceIdDoNotCollide`; operator-level: `ActionStateDivergenceRecoveryTest#normalFanoutWithIdenticalSiblingAttributesStillExecutesBothSiblings` | | 2. Replay-stable root identity | `ActionStateUtilTest#testDeriveRootEventOccurrenceIdIsReplayStableAndKeySensitive` | | 3. Content-divergence preserved | `ActionStateUtilTest#testOccurrenceIdDoesNotSuppressContentDivergence`; `ActionStateDivergenceRecoveryTest#changedInputHistoryMustInvalidateLaterResultAcrossTwoRestores` (pre-existing, now passing again) | | 4. Key format unchanged | pre-existing `ActionStateUtilTest` key-format tests (`testParseKey*`, `testIsKeyRetained*`), all still green | | 5. Legacy fallback | `ActionStateUtilTest#testDistinctEventOccurrencesWithoutOccurrenceIdCollideLegacy` (pins the documented legacy behavior) | Coverage by risk: the identity-derivation functions and the operator wiring that assigns them are both covered directly; the two closest pre-existing regression risks (content-based divergence detection, and a hand-fabricated durable-state fixture in `KafkaActionStateStoreRecoveryTest` that bypassed the new assignment path) were caught by the existing suite and fixed. Not verified: the Fluss-backed store path specifically (Fluss integration tests do not run in this environment; the fix is store-agnostic and goes through the same `ActionStateKeyEncoder` both Kafka and Fluss already share, and `FlussActionStateStore` received the identical two-method addition as `KafkaActionStateStore`). <details> <summary>Implementation notes and supporting evidence</summary> `generateUUIDForEvent` combines `occurrenceId` and the attributes-JSON bytes with a `#` separator before hashing through `UUID.nameUUIDFromBytes`, so the segment stays a canonical 36-character UUID regardless of which branch runs. `KafkaActionStateStoreRecoveryTest#queuedActionUsesRecoveredDurableResultAfterRestore` fabricates pending durable state directly against the store, bypassing `ActionExecutionOperator`'s assignment path; it now assigns the same `businessKeyIdentity#seqNum` identity to its hand-built `InputEvent` that the operator would compute for its own, matching the real contract instead of relying on the old content-only derivation. </details> ### API `Event` gains a public, `@JsonIgnore` field (`occurrenceId`) with a getter/setter, excluded from `equals`/`hashCode` and from JSON serialization — no change to any existing serialized payload or to the cross-language event contract. `ActionStateStore` gains two new interface methods (`generateBusinessKeyIdentity`, `generateActionExecutionId`); the two shipped implementations (`KafkaActionStateStore`, `FlussActionStateStore`) and the test `InMemoryActionStateStore` all implement them as one-line delegations to their existing `ActionStateKeyEncoder`. Any third-party `ActionStateStore` implementation outside this repository would need to add these two methods — a source-incompatible interface change for external implementers, mitigated by both being thin delegations every implementation already has the pieces for. No durable action-state key format or size change; no migration needed. State persisted by a pre-fix version has no `occurrenceId` and recovers exactly as before. ### Documentation - [ ] `doc-needed` - [x] `doc-not-needed` - [ ] `doc-included` ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No Generated-by: Claude Code 2.1.314 (Claude Sonnet 5) -- 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]
