This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 069084082 [runtime] Fix action-state key collisions between event
occurrences with identical attributes (#1342)
069084082 is described below
commit 06908408218d80bc2b12cf1f4ee7d36f21c1ae05
Author: Ashfaq <[email protected]>
AuthorDate: Wed Nov 4 08:47:29 2026 +0530
[runtime] Fix action-state key collisions between event occurrences with
identical attributes (#1342)
Generated-by: Claude Code 2.1.316 (Claude Sonnet 5)
---
.../java/org/apache/flink/agents/api/Event.java | 28 ++++++
.../runtime/actionstate/ActionStateUtil.java | 88 ++++++++++++++++-
.../runtime/operator/ActionExecutionOperator.java | 45 +++++++++
.../ActionStateDivergenceRecoveryTest.java | 34 +++++++
.../runtime/actionstate/ActionStateUtilTest.java | 104 +++++++++++++++++++++
.../actionstate/InMemoryActionStateStore.java | 9 ++
.../KafkaActionStateStoreRecoveryTest.java | 15 +++
.../operator/ActionExecutionOperatorTest.java | 34 ++++++-
8 files changed, 352 insertions(+), 5 deletions(-)
diff --git a/api/src/main/java/org/apache/flink/agents/api/Event.java
b/api/src/main/java/org/apache/flink/agents/api/Event.java
index 1613ecffd..9c87f6e90 100644
--- a/api/src/main/java/org/apache/flink/agents/api/Event.java
+++ b/api/src/main/java/org/apache/flink/agents/api/Event.java
@@ -60,6 +60,17 @@ public class Event {
*/
private Long sourceTimestamp;
+ /**
+ * Runtime-internal lineage identity for this specific event occurrence.
Not part of the
+ * cross-language event contract and not {@link #id}: {@link #id} is a
fresh random value on
+ * every construction (including on replay of the same logical input),
while this field is
+ * assigned deterministically by the runtime from the occurrence's
position in the causal tree
+ * (business key and sequence number for a root event; producing action
execution plus output
+ * ordinal for an action's output event). It lets durable action-state
keys distinguish sibling
+ * event occurrences that carry identical attributes.
+ */
+ private String occurrenceId;
+
/** Unified event with user-defined type and attributes. */
public Event(String type, Map<String, Object> attributes) {
this(UUID.randomUUID(), type, attributes, new HashMap<>());
@@ -156,6 +167,22 @@ public class Event {
this.sourceTimestamp = timestamp;
}
+ /** Returns this occurrence's runtime-assigned lineage identity, or {@code
null} if unset. */
+ @JsonIgnore
+ public String getOccurrenceId() {
+ return occurrenceId;
+ }
+
+ /**
+ * Assigns this occurrence's runtime-internal lineage identity. Called by
the runtime once, at
+ * the point the occurrence is created (root event ingestion or
action-output emission); not a
+ * user-facing API.
+ */
+ @JsonIgnore
+ public void setOccurrenceId(String occurrenceId) {
+ this.occurrenceId = occurrenceId;
+ }
+
/**
* Creates a base Event from another Event, copying its identity, data,
attachments, and
* framework metadata. Subclasses override this to reconstruct typed event
objects with proper
@@ -188,6 +215,7 @@ public class Event {
reconstructedEvent.attachments.clear();
reconstructedEvent.attachments.putAll(source.attachments);
reconstructedEvent.sourceTimestamp = source.sourceTimestamp;
+ reconstructedEvent.occurrenceId = source.occurrenceId;
return reconstructed;
}
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java
index e67553a73..cc8cf5949 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java
@@ -342,8 +342,92 @@ public final class ActionStateUtil {
}
private static String generateUUIDForEvent(Event event) throws IOException
{
- return String.valueOf(
-
UUID.nameUUIDFromBytes(MAPPER.writeValueAsBytes(event.getAttributes())));
+ byte[] attributesBytes =
MAPPER.writeValueAsBytes(event.getAttributes());
+ String occurrenceId = event.getOccurrenceId();
+ if (occurrenceId == null) {
+ // Attributes-only fallback. Distinct occurrences with identical
attributes collide
+ // here; kept only because a large pre-existing lower-level test
surface (key-encoding
+ // and store tests predating occurrence identity) builds events
without assigning one.
+ // Removing this fallback outright requires migrating that test
surface to assign an
+ // occurrence identity on every constructed event, tracked as a
separate follow-up.
+ return String.valueOf(UUID.nameUUIDFromBytes(attributesBytes));
+ }
+ // Folds the occurrence identity together with the existing attributes
hash, rather than
+ // replacing it: attributes must stay in the key so the content-based
replay-divergence
+ // protection this already provides (a later restore with a changed
input must not reuse a
+ // cached result from the other input's history) is unchanged, while
the occurrence
+ // identity additionally distinguishes sibling occurrences that carry
identical attributes,
+ // which attributes alone cannot. Folded through nameUUIDFromBytes so
the composite key's
+ // fixed five-segment, 36-char-UUID format is unchanged.
+ byte[] occurrenceBytes = occurrenceId.getBytes(StandardCharsets.UTF_8);
+ byte[] combined = new byte[occurrenceBytes.length + 1 +
attributesBytes.length];
+ System.arraycopy(occurrenceBytes, 0, combined, 0,
occurrenceBytes.length);
+ combined[occurrenceBytes.length] = '#';
+ System.arraycopy(
+ attributesBytes, 0, combined, occurrenceBytes.length + 1,
attributesBytes.length);
+ return String.valueOf(UUID.nameUUIDFromBytes(combined));
+ }
+
+ /**
+ * Derives the replay-stable occurrence identity of a root (input) event
from its business-key
+ * identity and sequence number. Stable across replay of the same logical
input: a replayed
+ * {@code InputEvent} gets a fresh random {@link Event#getId()} but the
same business key and
+ * sequence number, so it derives the same occurrence identity.
+ */
+ public static String deriveRootEventOccurrenceId(String
businessKeyIdentity, long seqNum) {
+ Preconditions.checkNotNull(businessKeyIdentity, "businessKeyIdentity
cannot be null.");
+ return sha256Base64(
+ (businessKeyIdentity + KEY_SEPARATOR +
seqNum).getBytes(StandardCharsets.UTF_8));
+ }
+
+ /**
+ * Derives the occurrence identity of one of an action execution's output
events from that
+ * execution's own action-execution identity (its composite action-state
key, which already
+ * folds in its own triggering event's occurrence identity and the
action's identity) and the
+ * event's 0-based position among that execution's emitted outputs. Two
sibling output events
+ * with identical attributes get distinct identities because their
ordinals differ; the same
+ * event re-derived during a deterministic replay gets the same identity
because the producing
+ * execution's identity and the emission order are themselves
replay-stable.
+ */
+ public static String deriveOutputEventOccurrenceId(
+ String producerActionExecutionId, int outputOrdinal) {
+ Preconditions.checkNotNull(
+ producerActionExecutionId, "producerActionExecutionId cannot
be null.");
+ Preconditions.checkArgument(
+ outputOrdinal >= 0, "outputOrdinal must be nonnegative but was
%s.", outputOrdinal);
+ return sha256Base64(
+ (producerActionExecutionId + KEY_SEPARATOR + outputOrdinal)
+ .getBytes(StandardCharsets.UTF_8));
+ }
+
+ /**
+ * Derives the identity of one action execution from its triggering
event's own occurrence
+ * identity and the action's identity. Independent of the business key,
sequence number, and
+ * durable storage: the business key and sequence number are already
folded into the triggering
+ * event's occurrence identity, so no store or key encoder is needed here.
+ *
+ * <p>Chat and internal-subagent call paths construct their own triggering
events (subgraph
+ * delegates, chat requests) without routing them through the root-event
assignment that gives
+ * an ordinary {@code InputEvent} its occurrence identity, so {@code
triggeringEvent} can arrive
+ * here with none set. Falls back to an attributes-only identity for that
event in that case,
+ * same tradeoff as {@link #generateUUIDForEvent}'s fallback: collides on
distinct occurrences
+ * with identical attributes, kept only because propagating occurrence
identity through those
+ * paths is a larger migration tracked as a separate follow-up.
+ */
+ public static String deriveActionExecutionId(Event triggeringEvent, Action
action)
+ throws IOException {
+ Preconditions.checkNotNull(triggeringEvent, "triggeringEvent cannot be
null.");
+ Preconditions.checkNotNull(action, "action cannot be null.");
+ String occurrenceId = triggeringEvent.getOccurrenceId();
+ String effectiveId =
+ occurrenceId != null
+ ? occurrenceId
+ : String.valueOf(
+ UUID.nameUUIDFromBytes(
+
MAPPER.writeValueAsBytes(triggeringEvent.getAttributes())));
+ return sha256Base64(
+ (effectiveId + KEY_SEPARATOR + generateUUIDForAction(action))
+ .getBytes(StandardCharsets.UTF_8));
}
private static String generateUUIDForAction(Action action) throws
IOException {
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
index a5bc2e4ef..23f0b17a9 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
@@ -36,7 +36,9 @@ import
org.apache.flink.agents.plan.subagent.InternalSubagentPlans;
import org.apache.flink.agents.plan.utils.CancellationUtils;
import org.apache.flink.agents.runtime.ResourceCache;
import org.apache.flink.agents.runtime.actionstate.ActionState;
+import org.apache.flink.agents.runtime.actionstate.ActionStateKeyEncoder;
import org.apache.flink.agents.runtime.actionstate.ActionStateStore;
+import org.apache.flink.agents.runtime.actionstate.ActionStateUtil;
import org.apache.flink.agents.runtime.async.ContinuationActionExecutor;
import org.apache.flink.agents.runtime.async.ContinuationContext;
import org.apache.flink.agents.runtime.context.RunnerContextImpl;
@@ -293,6 +295,11 @@ public class ActionExecutionOperator<IN, OUT> extends
AbstractStreamOperator<OUT
private final transient DurableExecutionManager durableExecManager;
+ // Derives root-event business-key identities directly from the operator's
own keyed-state
+ // serializer, independent of whether durable execution (and so an
ActionStateStore) is
+ // configured, so every root event gets a stable occurrence identity
regardless.
+ private transient ActionStateKeyEncoder occurrenceIdentityKeyEncoder;
+
private transient OperatorStateManager stateManager;
private transient TypeSerializer<Event> eventSerializer;
@@ -378,6 +385,8 @@ public class ActionExecutionOperator<IN, OUT> extends
AbstractStreamOperator<OUT
eventRouter.open(builtInMetrics);
int maxParallelism =
getRuntimeContext().getTaskInfo().getMaxNumberOfParallelSubtasks();
+ occurrenceIdentityKeyEncoder =
+ new ActionStateKeyEncoder(maxParallelism,
getActionStateKeySerializer());
durableExecManager.maybeInitActionStateStore(
agentPlan.getConfig(), maxParallelism,
getActionStateKeySerializer());
durableExecManager.initRecoveryMarkerState(getOperatorStateBackend());
@@ -616,6 +625,37 @@ public class ActionExecutionOperator<IN, OUT> extends
AbstractStreamOperator<OUT
}
}
+ /**
+ * Assigns a root (input) event's replay-stable occurrence identity,
derived from its business
+ * key and sequence number. Assigned unconditionally, independent of
whether durable execution
+ * is enabled, so chat and internal-subagent call paths can rely on it too.
+ */
+ private void assignRootOccurrenceId(Object key, Event event) throws
Exception {
+ String businessKeyIdentity =
occurrenceIdentityKeyEncoder.generateBusinessKeyIdentity(key);
+ event.setOccurrenceId(
+ ActionStateUtil.deriveRootEventOccurrenceId(
+ businessKeyIdentity,
stateManager.getSequenceNumber()));
+ }
+
+ /**
+ * Assigns each of an action execution's output events its lineage-based
occurrence identity
+ * before dispatch, so that distinct event occurrences with identical
attributes never collide
+ * in the durable action-state key. Derived purely from the triggering
event's own occurrence
+ * identity and the action's identity, so it needs neither the business
key nor a durable store.
+ */
+ private void assignOutputOccurrenceIds(ActionTask actionTask, List<Event>
outputEvents)
+ throws Exception {
+ String producerExecutionId =
+ ActionStateUtil.deriveActionExecutionId(actionTask.event,
actionTask.action);
+ for (int ordinal = 0; ordinal < outputEvents.size(); ordinal++) {
+ outputEvents
+ .get(ordinal)
+ .setOccurrenceId(
+ ActionStateUtil.deriveOutputEventOccurrenceId(
+ producerExecutionId, ordinal));
+ }
+ }
+
/** Processes an Action's output Event with its producer and upstream
references populated. */
private void processActionOutputEvent(
Object key, String contextKey, Event outputEvent, ActionTask
actionTask)
@@ -686,6 +726,7 @@ public class ActionExecutionOperator<IN, OUT> extends
AbstractStreamOperator<OUT
stateManager.addProcessingKey(key);
recordPriorities.put(key, new
RecordPriority(nextRecordIndex++));
stateManager.initOrIncSequenceNumber();
+ assignRootOccurrenceId(key, event);
if (freshRecordRound) {
// Announced once the record's own bookkeeping is in place
and before anything
// it triggers, so a listener never sees this record's
action callbacks outside
@@ -1037,6 +1078,7 @@ public class ActionExecutionOperator<IN, OUT> extends
AbstractStreamOperator<OUT
}
}
+ assignOutputOccurrenceIds(actionTask, outputEvents);
for (Event actionOutputEvent : outputEvents) {
processActionOutputEvent(key, contextKey, actionOutputEvent,
actionTask);
}
@@ -1222,6 +1264,8 @@ public class ActionExecutionOperator<IN, OUT> extends
AbstractStreamOperator<OUT
+ " or set execution.batch-state-backend.enabled to
false.");
int maxParallelism =
getRuntimeContext().getTaskInfo().getMaxNumberOfParallelSubtasks();
+ occurrenceIdentityKeyEncoder =
+ new ActionStateKeyEncoder(maxParallelism,
getActionStateKeySerializer());
durableExecManager.maybeInitActionStateStore(
agentPlan.getConfig(), maxParallelism,
getActionStateKeySerializer());
@@ -2108,6 +2152,7 @@ public class ActionExecutionOperator<IN, OUT> extends
AbstractStreamOperator<OUT
// callback belongs to, which would otherwise get none.
notifyActionReused(actionTask);
}
+ assignOutputOccurrenceIds(actionTask,
result.getOutputEvents());
} catch (Throwable t) {
if (lifecyclePrepared && actionTask != null) {
notifyActionFailed(actionTask, t);
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateDivergenceRecoveryTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateDivergenceRecoveryTest.java
index 4ee976b71..172231c5b 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateDivergenceRecoveryTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateDivergenceRecoveryTest.java
@@ -97,6 +97,26 @@ public class ActionStateDivergenceRecoveryTest {
6);
}
+ /**
+ * FLINK-AGENTS-1084 regression guard: two sibling output events with
identical attributes (not
+ * merely two different ones, as above) must each trigger a genuinely
separate action execution
+ * of "child" rather than the second silently reusing the first's durable
state. Without the
+ * occurrence-identity fix this collapses to 4 total calls (1 "input" + 1
"child" per input)
+ * instead of 6, because the second "child" execution's state lookup would
find the first's
+ * just-persisted completed state under the same (then-colliding) key.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = {"kafka", "fluss", "kafka-parallel",
"fluss-parallel"})
+ void
normalFanoutWithIdenticalSiblingAttributesStillExecutesBothSiblings(String
backend)
+ throws Exception {
+ assertNormalRecovery(
+ backend,
+ Map.of(
+ "input", action("input", "emitTwoIdentical",
InputEvent.EVENT_TYPE),
+ "child", action("child", "count", "step")),
+ 6);
+ }
+
private void assertNormalRecovery(String backend, Map<String, Action>
actions, int expected)
throws Exception {
CALLS.set(0);
@@ -301,6 +321,20 @@ public class ActionStateDivergenceRecoveryTest {
ctx.sendEvent(new Event("step", Map.of("value", "second")));
}
+ /**
+ * Emits two sibling output events with IDENTICAL attributes
(FLINK-AGENTS-1084's reproduction
+ * shape). Before the occurrence-identity fix, both siblings' durable
action-state keys
+ * collided: the second child's pre-execution lookup found the first
child's just-persisted
+ * completed state and incorrectly skipped its own real execution,
replaying the first child's
+ * result instead. See {@link #emitTwo} for the distinct-attributes
sibling case this
+ * complements.
+ */
+ public static void emitTwoIdentical(Event event, RunnerContext ctx) {
+ CALLS.incrementAndGet();
+ ctx.sendEvent(new Event("step", Map.of("value", "same")));
+ ctx.sendEvent(new Event("step", Map.of("value", "same")));
+ }
+
public static void setOrRead(Event event, RunnerContext ctx) throws
Exception {
CALLS.incrementAndGet();
String value = (String) event.getAttr("input");
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java
index bab111d74..92a131662 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java
@@ -17,6 +17,7 @@
*/
package org.apache.flink.agents.runtime.actionstate;
+import org.apache.flink.agents.api.Event;
import org.apache.flink.agents.api.InputEvent;
import org.apache.flink.agents.plan.actions.Action;
import org.apache.flink.agents.runtime.subgraph.SubgraphEvent;
@@ -27,6 +28,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.UUID;
@@ -61,6 +63,108 @@ public class ActionStateUtilTest {
assertEquals(key1, key2);
}
+ /**
+ * Reproduces the collision reported in FLINK-AGENTS-1084: without an
assigned occurrence
+ * identity (the legacy fallback, e.g. state written before this fix), two
distinct event
+ * occurrences with identical attributes reaching the same action under
the same business key
+ * and sequence number produce the same durable action-state key.
+ */
+ @Test
+ public void testDistinctEventOccurrencesWithoutOccurrenceIdCollideLegacy()
throws Exception {
+ Object businessKey = "key";
+ Action action = new NoOpAction("work-action");
+ Event first = new Event("WorkItem", Map.of("value", "same"));
+ Event second = new Event("WorkItem", Map.of("value", "same"));
+
+ assertNotEquals(first.getId(), second.getId());
+
+ String firstKey = generateKey(businessKey, 1L, action, first,
MAX_PARALLELISM);
+ String secondKey = generateKey(businessKey, 1L, action, second,
MAX_PARALLELISM);
+
+ assertEquals(firstKey, secondKey);
+ }
+
+ /**
+ * The fix for FLINK-AGENTS-1084: once the runtime assigns each output
event its lineage-based
+ * occurrence identity (as {@code ActionExecutionOperator} does via {@link
+ * #deriveOutputEventOccurrenceId}), two sibling occurrences with
identical attributes no longer
+ * collide, because their output ordinals differ.
+ */
+ @Test
+ public void testDistinctEventOccurrencesWithOccurrenceIdDoNotCollide()
throws Exception {
+ Object businessKey = "key";
+ Action action = new NoOpAction("work-action");
+ Event first = new Event("WorkItem", Map.of("value", "same"));
+ Event second = new Event("WorkItem", Map.of("value", "same"));
+
+ String producerExecutionId = generateKey(businessKey, 1L, action,
first, MAX_PARALLELISM);
+ first.setOccurrenceId(
+
ActionStateUtil.deriveOutputEventOccurrenceId(producerExecutionId, 0));
+ second.setOccurrenceId(
+
ActionStateUtil.deriveOutputEventOccurrenceId(producerExecutionId, 1));
+
+ String firstKey = generateKey(businessKey, 1L, action, first,
MAX_PARALLELISM);
+ String secondKey = generateKey(businessKey, 1L, action, second,
MAX_PARALLELISM);
+
+ assertNotEquals(firstKey, secondKey);
+ }
+
+ /**
+ * Content-based replay-divergence protection must survive the
occurrence-identity fix: a
+ * changed attribute still changes the key even when both occurrences
share the same lineage
+ * position (same producer execution and ordinal), as it would on a
genuine replay divergence.
+ */
+ @Test
+ public void testOccurrenceIdDoesNotSuppressContentDivergence() throws
Exception {
+ Object businessKey = "key";
+ Action action = new NoOpAction("work-action");
+ Event original = new Event("WorkItem", Map.of("value", "A"));
+ Event diverged = new Event("WorkItem", Map.of("value", "B"));
+ String sameLineage = "producer-exec-id#0";
+ original.setOccurrenceId(sameLineage);
+ diverged.setOccurrenceId(sameLineage);
+
+ String originalKey = generateKey(businessKey, 1L, action, original,
MAX_PARALLELISM);
+ String divergedKey = generateKey(businessKey, 1L, action, diverged,
MAX_PARALLELISM);
+
+ assertNotEquals(originalKey, divergedKey);
+ }
+
+ @Test
+ public void testDeriveRootEventOccurrenceIdIsReplayStableAndKeySensitive()
{
+ String a =
ActionStateUtil.deriveRootEventOccurrenceId("business-key-identity", 1L);
+ String b =
ActionStateUtil.deriveRootEventOccurrenceId("business-key-identity", 1L);
+ String differentSeqNum =
+
ActionStateUtil.deriveRootEventOccurrenceId("business-key-identity", 2L);
+ String differentKey =
ActionStateUtil.deriveRootEventOccurrenceId("other-identity", 1L);
+
+ // Replay of the same logical input (same business key, same sequence
number) is stable,
+ // independent of the fresh random Event#getId() each replay
constructs.
+ assertEquals(a, b);
+ assertNotEquals(a, differentSeqNum);
+ assertNotEquals(a, differentKey);
+ }
+
+ @Test
+ public void
testDeriveOutputEventOccurrenceIdIsOrdinalAndProducerSensitive() {
+ String producer = "producer-execution-id";
+ String ordinal0 =
ActionStateUtil.deriveOutputEventOccurrenceId(producer, 0);
+ String ordinal0Again =
ActionStateUtil.deriveOutputEventOccurrenceId(producer, 0);
+ String ordinal1 =
ActionStateUtil.deriveOutputEventOccurrenceId(producer, 1);
+ String otherProducer =
ActionStateUtil.deriveOutputEventOccurrenceId("other-producer", 0);
+
+ assertEquals(ordinal0, ordinal0Again);
+ assertNotEquals(ordinal0, ordinal1);
+ assertNotEquals(ordinal0, otherProducer);
+ }
+
+ @Test
+ public void testDeriveOutputEventOccurrenceIdRejectsNegativeOrdinal() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
ActionStateUtil.deriveOutputEventOccurrenceId("producer", -1));
+ }
+
@ParameterizedTest
@ValueSource(booleans = {false, true})
void subgraphStateKeyIgnoresDelegateIdButSeparatesInvocations(boolean
bootstrap)
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/InMemoryActionStateStore.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/InMemoryActionStateStore.java
index 5a10e0d39..73f3e33b6 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/InMemoryActionStateStore.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/InMemoryActionStateStore.java
@@ -70,6 +70,15 @@ public class InMemoryActionStateStore implements
ActionStateStore {
.get(keyEncoder.generateKey(key, seqNum, action, event));
}
+ /**
+ * Test-only convenience accessor to this store's own business-key
identity, matching what the
+ * runtime now derives independently via its own {@link
ActionStateKeyEncoder} (not part of the
+ * {@link ActionStateStore} contract).
+ */
+ public String generateBusinessKeyIdentity(Object key) {
+ return keyEncoder.generateBusinessKeyIdentity(key);
+ }
+
@Override
public void rebuildState(List<Object> recoveryMarker) {
// No-op for in-memory store as it does not persist state;
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreRecoveryTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreRecoveryTest.java
index dd85cc3af..215cd468a 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreRecoveryTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreRecoveryTest.java
@@ -31,6 +31,7 @@ import org.apache.flink.agents.plan.actions.Action;
import org.apache.flink.agents.runtime.operator.ActionExecutionOperator;
import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorFactory;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -165,6 +166,20 @@ public class KafkaActionStateStoreRecoveryTest {
harness.getOutput().clear();
InputEvent inputEvent = new InputEvent(7L);
+ // The real pipeline assigns a root occurrence identity
(businessKeyIdentity#seqNum)
+ // to every InputEvent before it reaches the durable store; this
fixture fabricates
+ // pending state directly, bypassing that assignment, so it must
assign the same
+ // identity itself or the key this computes below would not match
the key the operator
+ // computes for its own InputEvent(7L) on the same (key=1L,
seqNum=1). The operator
+ // derives this identity from the keyed-state backend's own
serializer (LongSerializer
+ // for this Long-keyed harness), not from the store's encoder
(which uses a generic
+ // Object serializer in this test suite, per
ActionStateTestUtils.KEY_SERIALIZER), so
+ // this must match the former, not
createKeyEncoder(MAX_PARALLELISM).
+ inputEvent.setOccurrenceId(
+ ActionStateUtil.deriveRootEventOccurrenceId(
+ ActionStateUtil.generateBusinessKeyIdentity(
+ 1L, LongSerializer.INSTANCE),
+ 1L));
ActionState pendingState = new ActionState(inputEvent);
pendingState.addCallResult(
new CallResult("multiply",
OBJECT_MAPPER.writeValueAsBytes(callExternal(7L))));
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java
index 54ad6918c..05ea0de9c 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java
@@ -86,6 +86,7 @@ import
org.apache.flink.agents.runtime.tracelog.TraceLogWriter;
import org.apache.flink.api.common.serialization.SerializerConfigImpl;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
@@ -1677,12 +1678,18 @@ public class ActionExecutionOperatorTest {
(List<StreamRecord<Object>>) testHarness.getRecordOutput();
assertThat(output).singleElement().extracting(StreamRecord::getValue).isEqualTo(7L);
+ InputEvent lookupEvent = new InputEvent(7L);
+ lookupEvent.setOccurrenceId(
+ ActionStateUtil.deriveRootEventOccurrenceId(
+ ActionStateUtil.generateBusinessKeyIdentity(
+ "7", StringSerializer.INSTANCE),
+ 0L));
ActionState actionState =
actionStateStore.get(
"7",
0L,
agentPlan.getActions().get("bestEffortMemoryObservationAction"),
- new InputEvent(7L));
+ lookupEvent);
assertThat(actionState).isNotNull();
assertThat(actionState.getOutputEvents())
.filteredOn(ShortTermWriteEvent.class::isInstance)
@@ -5380,7 +5387,7 @@ public class ActionExecutionOperatorTest {
String actionName,
ActionState actionState)
throws Exception {
- InputEvent event = new InputEvent(input);
+ InputEvent event = rootInputEventWithOccurrenceId(key, input);
Action action = agentPlan.getActions().get(actionName);
actionStateStore.put(key, 0L, action, event, actionState);
}
@@ -5392,11 +5399,32 @@ public class ActionExecutionOperatorTest {
AgentPlan agentPlan,
String actionName)
throws Exception {
- InputEvent event = new InputEvent(input);
+ InputEvent event = rootInputEventWithOccurrenceId(key, input);
Action action = agentPlan.getActions().get(actionName);
return actionStateStore.get(key, 0L, action, event);
}
+ /**
+ * Builds an {@link InputEvent} carrying the same root occurrence identity
the operator itself
+ * assigns on the first (seqNum=0) event for this key, so a fixture's
directly-seeded state keys
+ * to the same durable-store key the operator computes when it processes
its own {@code
+ * InputEvent} for this key. Uses {@code LongSerializer} directly (the
real keyed-state
+ * backend's serializer for these {@code Long}-keyed harnesses, per {@link
+ * #testDefaultStoreUsesKeyedStateBackendSerializer}) rather than the
store's own encoder: the
+ * operator now derives this identity independent of the store, and a
store built standalone
+ * with a generic serializer (as these fixtures' {@link
InMemoryActionStateStore} instances are)
+ * would otherwise compute a different, mismatched identity.
+ */
+ private static InputEvent rootInputEventWithOccurrenceId(long key, long
input)
+ throws Exception {
+ InputEvent event = new InputEvent(input);
+ event.setOccurrenceId(
+ ActionStateUtil.deriveRootEventOccurrenceId(
+ ActionStateUtil.generateBusinessKeyIdentity(key,
LongSerializer.INSTANCE),
+ 0L));
+ return event;
+ }
+
/**
* Records the ownership filter that {@code
DurableExecutionManager.handleRecovery} installs on
* the store during operator recovery, so tests can assert which
key-groups a given subtask
