wenjin272 commented on code in PR #1342:
URL: https://github.com/apache/flink-agents/pull/1342#discussion_r4452644507
##########
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:
Chat/internal-subagent call IDs were not part of the original issue scope,
but we later identified the same event-occurrence identity problem in #1220. I
suggest fixing that here as well; a follow-up PR is also fine. Either way,
please move identity generation into a shared runtime component in this PR so
both paths can use it independently of durable storage.
For the chat/subagent fix, `DeterministicIdAllocator.Namespace` and its
caller setup in `ActionTaskContextManager` should consume the shared action
execution identity while retaining target/session/call scoping. The current
allocator does not use occurrence identity, so identical event occurrences can
still produce duplicate IDs in `SubgraphCallManager`, including when durable
execution is disabled.
--
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]
