wenjin272 commented on code in PR #1342:
URL: https://github.com/apache/flink-agents/pull/1342#discussion_r4452644496
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java:
##########
@@ -69,6 +69,30 @@ void put(Object key, long seqNum, Action action, Event
event, ActionState state)
*/
ActionState get(Object key, long seqNum, Action action, Event event)
throws Exception;
+ /**
+ * Returns the business-key identity segment this store's key encoder
would embed in a composite
+ * action-state key for {@code key}. Used by the runtime to derive a
replay-stable occurrence
+ * identity for root events (FLINK-AGENTS-1084); unrelated to storing or
retrieving state.
+ *
+ * @param key the key associated with the message
+ */
+ String generateBusinessKeyIdentity(Object key);
+
+ /**
+ * Returns the full composite action-state key this store's key encoder
would use for {@code
+ * (key, seqNum, action, event)} — the deterministic identity of one
action execution. Used by
+ * the runtime to derive occurrence identities for that execution's output
events
+ * (FLINK-AGENTS-1084); unrelated to storing or retrieving state.
+ *
+ * @param key the key associated with the message
+ * @param seqNum the sequence number of the key
+ * @param action the action the agent is taking
+ * @param event the event that triggered the action
+ * @throws IOException when key generation failed
+ */
+ String generateActionExecutionId(Object key, long seqNum, Action action,
Event event)
Review Comment:
Could we move identity generation into a shared runtime component instead of
adding these methods to `ActionStateStore`? Kafka and Fluss only delegate to
the same encoder, and execution identity should be available independently of
durable storage.
An action execution ID can be derived as `hash(triggeringEventOccurrenceId,
actionIdentity)`, with business key and sequence number already represented in
the root occurrence ID. Content-based divergence checks should remain in place
separately.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java:
##########
@@ -342,8 +342,62 @@ private static String describeKey(@Nullable String
stateKey) {
}
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) {
+ // Legacy fallback for events the runtime has not assigned an
occurrence identity to
+ // (for example, state written before this fix by an older
version). Distinct
+ // occurrences with identical attributes still collide here; this
path exists only for
+ // mixed-version recovery continuity, not as the steady-state
behavior.
+ return String.valueOf(UUID.nameUUIDFromBytes(attributesBytes));
Review Comment:
We do not require compatibility with old persisted state for this change, so
please remove the legacy attributes-only fallback and make missing occurrence
identity explicit on paths that require it. Keeping this fallback preserves the
collision behavior the change is intended to eliminate.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -616,6 +617,44 @@ private void processInputEvent(Object key, Event
inputEvent) throws Exception {
}
}
+ /**
+ * Assigns a root (input) event's replay-stable occurrence identity,
derived from its business
+ * key and sequence number. No-op when durable execution is disabled,
since nothing keys action
+ * state on it in that case. See FLINK-AGENTS-1084.
+ */
+ private void assignRootOccurrenceId(Object key, Event event) throws
Exception {
+ String businessKeyIdentity =
durableExecManager.generateBusinessKeyIdentity(key);
+ if (businessKeyIdentity != null) {
+ 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. No-op when durable execution is
disabled. See
+ * FLINK-AGENTS-1084.
+ */
+ private void assignOutputOccurrenceIds(
+ Object key, long sequenceNumber, ActionTask actionTask,
List<Event> outputEvents)
+ throws Exception {
+ String producerExecutionId =
+ durableExecManager.generateActionExecutionId(
+ key, sequenceNumber, actionTask.action,
actionTask.event);
+ if (producerExecutionId == null) {
+ return;
Review Comment:
Please also wire the occurrence identity into chat/internal-subagent call
IDs, including when durable execution is disabled. This was part of the scope
discussed in #1084 and #1220.
Specifically, `DeterministicIdAllocator.Namespace` still derives IDs from
key, sequence number, action name, event type/attributes, and target name
without using occurrence identity. Please update it and its caller setup in
`ActionTaskContextManager` to consume the shared action execution identity,
while retaining target/session/call scoping. Otherwise, two identical event
occurrences can still produce the same call ID and fail registration in
`SubgraphCallManager`.
##########
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. See FLINK-AGENTS
issue #1084.
Review Comment:
Could you clean up the comments throughout this patch to describe the
current identity and recovery contracts directly? Repeated issue references
such as `FLINK-AGENTS-1084` and historical wording like “before this fix” are
unnecessary in production code; that context belongs in the PR description or
commit message.
--
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]
