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 982d71fb4 [api][runtime][java][python] Reject framework-owned lineage
at event emission (#1212)
982d71fb4 is described below
commit 982d71fb47e183f170111ac9eeac5d10eb0d5f20
Author: Sangkyoon Nam <[email protected]>
AuthorDate: Fri Oct 9 15:59:19 2026 +0900
[api][runtime][java][python] Reject framework-owned lineage at event
emission (#1212)
Generated-by: Claude Code 2.1.295 (Claude Opus 5.5)
---
.../java/org/apache/flink/agents/api/Event.java | 11 +-
.../flink/agents/api/context/RunnerContext.java | 4 +-
docs/content/docs/development/workflow_agent.md | 9 +-
python/flink_agents/api/events/event.py | 20 +++-
python/flink_agents/api/runner_context.py | 7 +-
python/flink_agents/api/tests/test_event.py | 4 +-
.../flink_agents/runtime/flink_runner_context.py | 17 +++
.../runtime/tests/test_flink_runner_context.py | 60 +++++++++-
.../agents/runtime/context/RunnerContextImpl.java | 29 +++++
.../runtime/context/RunnerContextLineageTest.java | 127 +++++++++++++++++++++
.../agents/runtime/operator/ActionTaskTest.java | 19 +++
11 files changed, 294 insertions(+), 13 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 7a7024607..593bdae31 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
@@ -79,8 +79,9 @@ public class Event {
/**
* Reconstructs an Event with an existing identity and optional
framework-managed lineage.
*
- * <p>The lineage values support deserialization and reconstruction. When
an Action emits this
- * Event, the runtime overwrites them with the current trigger Event ID
and Action name.
+ * <p>The lineage values support deserialization and reconstruction. {@code
+ * RunnerContext#sendEvent} rejects an Event that carries them; the
runtime sets lineage to the
+ * current trigger Event ID and Action name when it finalizes the Action's
outputs.
*
* @param id the existing Event ID
* @param type the Event type used for routing
@@ -160,7 +161,8 @@ public class Event {
/**
* Sets the framework-managed ID of the Event consumed by the emitting
Action.
*
- * <p>The runtime overwrites this value when an Action emits the Event.
+ * <p>The runtime sets this value when it finalizes an Action's outputs.
{@code
+ * RunnerContext#sendEvent} rejects an Event that already carries it.
*/
public void setUpstreamEventId(@Nullable UUID upstreamEventId) {
this.upstreamEventId = upstreamEventId;
@@ -176,7 +178,8 @@ public class Event {
/**
* Sets the framework-managed name of the Action that emitted this Event.
*
- * <p>The runtime overwrites this value when an Action emits the Event.
+ * <p>The runtime sets this value when it finalizes an Action's outputs.
{@code
+ * RunnerContext#sendEvent} rejects an Event that already carries it.
*/
public void setUpstreamActionName(@Nullable String upstreamActionName) {
this.upstreamActionName = upstreamActionName;
diff --git
a/api/src/main/java/org/apache/flink/agents/api/context/RunnerContext.java
b/api/src/main/java/org/apache/flink/agents/api/context/RunnerContext.java
index 08bf79ce5..c1ef17e20 100644
--- a/api/src/main/java/org/apache/flink/agents/api/context/RunnerContext.java
+++ b/api/src/main/java/org/apache/flink/agents/api/context/RunnerContext.java
@@ -36,7 +36,9 @@ public interface RunnerContext {
/**
* Sends an event.
*
- * @param event the event to be sent
+ * @param event the event to be sent; a new Event without upstream lineage
+ * @throws IllegalArgumentException if the event already carries {@code
upstreamEventId} or
+ * {@code upstreamActionName}, which the runtime sets when the Action
emits it
*/
void sendEvent(Event event);
diff --git a/docs/content/docs/development/workflow_agent.md
b/docs/content/docs/development/workflow_agent.md
index 48f1a6243..9fbeebcbc 100644
--- a/docs/content/docs/development/workflow_agent.md
+++ b/docs/content/docs/development/workflow_agent.md
@@ -773,9 +773,12 @@ public static void handleMyEvent(Event event,
RunnerContext ctx) {
{{< hint info >}}
`upstreamEventId` and `upstreamActionName` (`upstream_event_id` and
-`upstream_action_name` in Python) are framework-managed lineage metadata. User
code should keep
-user data in `attributes`. Values accepted during deserialization or
reconstruction are
-overwritten when an Action emits the Event.
+`upstream_action_name` in Python) are framework-managed lineage metadata, set
when an Action
+emits the Event. User code should keep user data in `attributes`. `sendEvent`
/ `send_event`
+rejects an Event that already carries lineage, such as an Event received from
another Action or
+a typed reconstruction of one: emit a new Event instead. Deserialization and
typed
+reconstruction keep the lineage an Event already has; after recovery, restored
outputs are bound
+to the replayed trigger again.
{{< /hint >}}
### JSON Serialization
diff --git a/python/flink_agents/api/events/event.py
b/python/flink_agents/api/events/event.py
index 2150f3235..725c6d736 100644
--- a/python/flink_agents/api/events/event.py
+++ b/python/flink_agents/api/events/event.py
@@ -31,11 +31,12 @@ from pydantic import (
BaseModel,
Field,
SerializerFunctionWrapHandler,
+ ValidationError,
field_validator,
model_serializer,
model_validator,
)
-from pydantic_core import PydanticSerializationError
+from pydantic_core import PydanticCustomError, PydanticSerializationError
from pyflink.common import Row
from flink_agents.api.memory_reference import MemoryRef
@@ -229,6 +230,23 @@ class Event(BaseModel, extra="allow"):
return self
def __setattr__(self, name: str, value: Any) -> None:
+ # Pydantic's own frozen-field error would not say how to keep an
identity.
+ if name == "id":
+ raise ValidationError.from_exception_data(
+ type(self).__name__,
+ [
+ {
+ "type": PydanticCustomError(
+ "frozen_event_id",
+ "Field is frozen. An Event's id is fixed when it
is "
+ "created; to give a typed Event the identity of an
"
+ "existing Event, use typed =
typed.reconstruct_from(existing).",
+ ),
+ "loc": ("id",),
+ "input": value,
+ }
+ ],
+ )
super().__setattr__(name, value)
# Raw attachments are offloaded to sensory memory before sending.
Validate every
# other field here without serializing those payloads.
diff --git a/python/flink_agents/api/runner_context.py
b/python/flink_agents/api/runner_context.py
index d14e171a7..254035649 100644
--- a/python/flink_agents/api/runner_context.py
+++ b/python/flink_agents/api/runner_context.py
@@ -143,7 +143,12 @@ class RunnerContext(ABC):
Parameters
----------
event : Event
- The event to be sent.
+ The event to be sent; a new Event without upstream lineage.
+
+ Raises:
+ ValueError: If the event already carries ``upstream_event_id`` or
+ ``upstream_action_name``, which the runtime sets when the
Action
+ emits it.
"""
@abstractmethod
diff --git a/python/flink_agents/api/tests/test_event.py
b/python/flink_agents/api/tests/test_event.py
index ce9454614..5ea521534 100644
--- a/python/flink_agents/api/tests/test_event.py
+++ b/python/flink_agents/api/tests/test_event.py
@@ -124,9 +124,11 @@ def test_event_id_does_not_change_with_event_content() ->
None:
def test_event_id_cannot_be_reassigned() -> None:
event = Event(type="test")
- with pytest.raises(ValidationError, match="Field is frozen"):
+ with pytest.raises(ValidationError, match="Field is frozen") as raised:
event.id = uuid4()
+ assert "reconstruct_from" in str(raised.value)
+
def test_explicit_none_id_generates_uuid() -> None:
event = Event(id=None, type="x")
diff --git a/python/flink_agents/runtime/flink_runner_context.py
b/python/flink_agents/runtime/flink_runner_context.py
index 140d5b29c..16cd59b95 100644
--- a/python/flink_agents/runtime/flink_runner_context.py
+++ b/python/flink_agents/runtime/flink_runner_context.py
@@ -600,6 +600,23 @@ class FlinkRunnerContext(RunnerContext, ExecutionReporter):
event : Event
The event to be processed by the agent system.
"""
+ preset = [
+ f"{name}={value}"
+ for name, value in (
+ ("upstream_event_id", event.upstream_event_id),
+ ("upstream_action_name", event.upstream_action_name),
+ )
+ if value is not None
+ ]
+ if preset:
+ # Java's sendEvent checks this too; checking here names the Python
fields.
+ err_msg = (
+ f"Event '{event.get_type()}' ({event.id}) already carries "
+ f"{' and '.join(preset)}. The runtime sets lineage when an
Action "
+ "emits an Event: emit a new Event rather than one received
from "
+ "another Action, and keep user metadata in attributes."
+ )
+ raise ValueError(err_msg)
store_event_attachments(event, self)
event_json = event.model_dump_json()
try:
diff --git a/python/flink_agents/runtime/tests/test_flink_runner_context.py
b/python/flink_agents/runtime/tests/test_flink_runner_context.py
index f283bab02..2a4562632 100644
--- a/python/flink_agents/runtime/tests/test_flink_runner_context.py
+++ b/python/flink_agents/runtime/tests/test_flink_runner_context.py
@@ -16,8 +16,11 @@
# limitations under the License.
#################################################################################
from typing import Any
+from uuid import UUID
-from flink_agents.api.events.event import Event
+import pytest
+
+from flink_agents.api.events.event import Event, InputEvent
from flink_agents.api.memory_reference import MemoryRef
from flink_agents.runtime.flink_runner_context import FlinkRunnerContext
@@ -51,10 +54,15 @@ class _FakeJavaRunnerContext:
self.sent_event_json = event_json
-def test_send_event_offloads_attachments_before_forwarding() -> None:
+def _send_event_context() -> tuple[FlinkRunnerContext, _FakeJavaRunnerContext]:
java_context = _FakeJavaRunnerContext()
ctx = FlinkRunnerContext.__new__(FlinkRunnerContext)
ctx._j_runner_context = java_context
+ return ctx, java_context
+
+
+def test_send_event_offloads_attachments_before_forwarding() -> None:
+ ctx, java_context = _send_event_context()
payload = {"value": "original"}
event = Event(type="AttachmentStep", attachments={"payload": payload})
@@ -66,3 +74,51 @@ def test_send_event_offloads_attachments_before_forwarding()
-> None:
assert java_context.sent_event_json is not None
forwarded_event = Event.from_json(java_context.sent_event_json)
assert forwarded_event.get_attachment("payload") == attachment
+
+
[email protected](
+ ("field", "value"),
+ [
+ ("upstream_event_id", UUID("00000000-0000-0000-0000-000000000001")),
+ ("upstream_action_name", "user_action"),
+ ],
+)
+def test_send_event_rejects_preset_lineage(field: str, value: Any) -> None:
+ ctx, java_context = _send_event_context()
+ event = Event(type="result", **{field: value})
+
+ with pytest.raises(ValueError, match=f"carries {field}=") as raised:
+ ctx.send_event(event)
+
+ other = ({"upstream_event_id", "upstream_action_name"} - {field}).pop()
+ assert f"{other}=" not in str(raised.value)
+
+ assert java_context.sent_event_json is None
+
+
+def test_send_event_rejects_reconstructed_event_with_lineage() -> None:
+ ctx, java_context = _send_event_context()
+ received = Event(
+ type="generic",
+ attributes={"input": 1},
+ upstream_event_id=UUID("00000000-0000-0000-0000-000000000002"),
+ upstream_action_name="upstream_action",
+ )
+ reconstructed = InputEvent(input=1).reconstruct_from(received)
+
+ with pytest.raises(ValueError, match="new Event"):
+ ctx.send_event(reconstructed)
+
+ assert java_context.sent_event_json is None
+
+
+def test_send_event_forwards_fresh_event_without_lineage() -> None:
+ ctx, java_context = _send_event_context()
+
+ ctx.send_event(Event(type="result", attributes={"source": "user"}))
+
+ assert java_context.sent_event_json is not None
+ forwarded = Event.from_json(java_context.sent_event_json)
+ assert forwarded.upstream_event_id is None
+ assert forwarded.upstream_action_name is None
+ assert forwarded.get_attr("source") == "user"
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
index 691be4113..fdde4ed7e 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
@@ -218,6 +218,7 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
@Override
public void sendEvent(Event event) {
mailboxThreadChecker.run();
+ checkNoPresetLineage(event);
try {
EventAttachmentUtils.storeEventAttachments(event, this);
} catch (Exception e) {
@@ -233,6 +234,34 @@ public class RunnerContextImpl implements RunnerContext,
ExecutionReporter {
pendingEvents.add(event);
}
+ /**
+ * Rejects an emitted Event that already carries lineage. The framework
binds lineage to the
+ * emitting Action when it finalizes the Action's outputs; outputs
restored from action state
+ * reach that step without passing through here.
+ */
+ private static void checkNoPresetLineage(Event event) {
+ List<String> preset = new ArrayList<>();
+ if (event.getUpstreamEventId() != null) {
+ preset.add("upstreamEventId=" + event.getUpstreamEventId());
+ }
+ if (event.getUpstreamActionName() != null) {
+ preset.add("upstreamActionName=" + event.getUpstreamActionName());
+ }
+ if (preset.isEmpty()) {
+ return;
+ }
+ throw new IllegalArgumentException(
+ "Event '"
+ + event.getType()
+ + "' ("
+ + event.getId()
+ + ") already carries "
+ + String.join(" and ", preset)
+ + ". The runtime sets lineage when an Action emits an
Event: emit a new"
+ + " Event rather than one received from another
Action, and keep user"
+ + " metadata in attributes.");
+ }
+
public List<Event> drainEvents(Long timestamp) {
mailboxThreadChecker.run();
return drainPendingEvents(timestamp);
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextLineageTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextLineageTest.java
new file mode 100644
index 000000000..6a6f14fd5
--- /dev/null
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/context/RunnerContextLineageTest.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.context;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.InputEvent;
+import org.apache.flink.agents.plan.AgentPlan;
+import org.apache.flink.agents.runtime.metrics.FlinkAgentsMetricGroupImpl;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests that emitted Events cannot carry framework-owned lineage. */
+class RunnerContextLineageTest {
+
+ @Test
+ void freshEventIsAccepted() {
+ RunnerContextImpl context = newContext();
+ Event event = new Event("result");
+
+ context.sendEvent(event);
+
+ assertThat(context.drainEvents(null)).containsExactly(event);
+ }
+
+ @Test
+ void presetUpstreamEventIdIsRejected() {
+ RunnerContextImpl context = newContext();
+ Event event = new Event("result");
+ event.setUpstreamEventId(UUID.randomUUID());
+
+ assertThatThrownBy(() -> context.sendEvent(event))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("upstreamEventId=")
+ .hasMessageNotContaining("upstreamActionName=")
+ .hasMessageContaining("attributes");
+ assertThat(context.drainEvents(null)).isEmpty();
+ }
+
+ @Test
+ void presetUpstreamActionNameIsRejected() {
+ RunnerContextImpl context = newContext();
+ Event event = new Event("result");
+ event.setUpstreamActionName("user_action");
+
+ assertThatThrownBy(() -> context.sendEvent(event))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("upstreamActionName=user_action")
+ .hasMessageNotContaining("upstreamEventId=");
+ assertThat(context.drainEvents(null)).isEmpty();
+ }
+
+ @Test
+ void typedReconstructionOfAnEventFromAnotherActionIsRejected() {
+ RunnerContextImpl context = newContext();
+ Event received = new Event("generic", new HashMap<>());
+ received.setUpstreamEventId(UUID.randomUUID());
+ received.setUpstreamActionName("upstream_action");
+ received.setAttr("input", 1L);
+
+ Event reconstructed = InputEvent.fromEvent(received);
+
+
assertThat(reconstructed.getUpstreamEventId()).isEqualTo(received.getUpstreamEventId());
+ assertThatThrownBy(() -> context.sendEvent(reconstructed))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("new Event");
+ }
+
+ @Test
+ void lineageArrivingThroughJsonIsRejected() throws Exception {
+ // The Python bridge deserializes the event with Event.fromJson before
sendEvent.
+ RunnerContextImpl context = newContext();
+ Event event =
+ Event.fromJson(
+
"{\"type\":\"result\",\"attributes\":{},\"upstreamEventId\":\""
+ + UUID.randomUUID()
+ +
"\",\"upstreamActionName\":\"python_action\"}");
+
+ assertThatThrownBy(() -> context.sendEvent(event))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("python_action");
+ }
+
+ private static RunnerContextImpl newContext() {
+ RunnerContextImpl context =
+ new RunnerContextImpl(
+ new FlinkAgentsMetricGroupImpl(
+
UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup()),
+ () -> {},
+ new AgentPlan(new HashMap<>(), new HashMap<>()),
+ null,
+ "job");
+ List<Event> buffer = new ArrayList<>();
+ context.switchActionContext(
+ "action",
+ new RunnerContextImpl.MemoryContext(null, null),
+ buffer,
+ "key",
+ "obs",
+ false,
+ null);
+ return context;
+ }
+}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionTaskTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionTaskTest.java
index ea24d79ca..988a746e5 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionTaskTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionTaskTest.java
@@ -28,6 +28,7 @@ import
org.apache.flink.agents.runtime.python.utils.PythonActionExecutor;
import org.junit.jupiter.api.Test;
import java.util.List;
+import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -128,6 +129,24 @@ class ActionTaskTest {
assertThat(outputEvent.getUpstreamActionName()).isEqualTo(action.getName());
}
+ @Test
+ void resultRebindsLineageOfOutputsRestoredFromActionState() {
+ // Recovered outputs reach finalization without sendEvent, still
carrying the lineage
+ // bound before the failure; they keep identity and are bound to this
task again.
+ Event triggeringEvent = new InputEvent(1L);
+ Action action = TestActions.noopAction();
+ ActionTask task = new JavaActionTask("key", triggeringEvent, action,
1L);
+ Event restoredOutput = new Event("result");
+ restoredOutput.setUpstreamEventId(UUID.randomUUID());
+ restoredOutput.setUpstreamActionName("previous_action");
+
+ List<Event> outputs =
task.finalizeOutputEvents(List.of(restoredOutput));
+
+ assertThat(outputs).containsExactly(restoredOutput);
+
assertThat(restoredOutput.getUpstreamEventId()).isEqualTo(triggeringEvent.getId());
+
assertThat(restoredOutput.getUpstreamActionName()).isEqualTo(action.getName());
+ }
+
@Test
void resultRejectsSelfLoopBeforeMutatingAnyOutput() {
Event triggeringEvent = new InputEvent(1L);
