From e8328a3f1e9d3ec181312cb11d03c77088dded66 Mon Sep 17 00:00:00 2001 From: rosemaryYuan <91046107+rosemarYuan@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:50:24 +0800 Subject: [PATCH 1/3] [api][python][test] Align Event ID generation with Java --- .../docs/development/workflow_agent.md | 27 ++++----- python/flink_agents/api/events/chat_event.py | 6 +- .../api/events/context_retrieval_event.py | 6 +- python/flink_agents/api/events/event.py | 43 +++++--------- python/flink_agents/api/events/tool_event.py | 10 +--- .../test_cross_language_event_snapshots.py | 3 +- python/flink_agents/api/tests/test_event.py | 58 ++++++++++++++++++- .../flink_integration_agent.py | 3 +- .../long_term_memory_test.py | 3 +- .../runtime/tests/test_python_java_utils.py | 23 ++++++-- 10 files changed, 111 insertions(+), 71 deletions(-) diff --git a/docs/content/docs/development/workflow_agent.md b/docs/content/docs/development/workflow_agent.md index 74589a1fe..0540c239d 100644 --- a/docs/content/docs/development/workflow_agent.md +++ b/docs/content/docs/development/workflow_agent.md @@ -658,11 +658,8 @@ class MyEvent(Event): @override def from_event(cls, event: Event) -> "MyEvent": assert "value" in event.attributes - result = MyEvent(value=event.attributes["value"]) - # Preserve the base event id. Assign it last: the content-based id is - # regenerated whenever another field changes. - result.id = event.id - return result + result = cls(value=event.attributes["value"]) + return result.with_framework_metadata_from(event) @property def value(self) -> str: @@ -707,16 +704,14 @@ public class MyEvent extends Event { {{< /tabs >}} {{< hint info >}} -When reconstructing a typed event, preserve the base `Event` metadata so that event logs, -listeners, correlation, deduplication, and downstream timestamp propagation stay consistent with -built-in events: - -- **`id`**: copy the source event's `id` onto the reconstructed event, as all built-in events do - in both languages. In Python, assign `result.id = event.id` **last**, because the content-based - `id` is regenerated whenever any other field changes. -- **`sourceTimestamp`** (Java only): carry it over with `setSourceTimestamp(...)` when - `hasSourceTimestamp()` is true, matching built-in Java events. This field is runtime-internal and - used for timestamp propagation; the Python `Event` has no equivalent. +Typed reconstruction represents the same Event occurrence, so it must preserve the base Event's +identity and framework-managed metadata: + +- **Python**: return `with_framework_metadata_from(event)` after constructing the typed object. It + returns a new typed object with the Event's UUIDv4 `id`; the returned Event's `id` remains + immutable. +- **Java**: pass `event.getId()` to the typed constructor, then carry over `sourceTimestamp` with + `setSourceTimestamp(...)` when `hasSourceTimestamp()` is true. {{< /hint >}} {{< hint info >}} @@ -756,4 +751,4 @@ private static Map normalizeAttributes(Map attri There are several built-in `Event` and `Action` in Flink-Agents: * See [Chat Models]({{< ref "docs/development/chat_models#built-in-events-and-actions" >}}) for how to chat with a LLM leveraging built-in action and events. * See [Tool Use]({{< ref "docs/development/tool_use#built-in-events-and-actions" >}}) for how to programmatically use a tool leveraging built-in action and events. -* See [Vector Stores]({{< ref "docs/development/vector_stores#built-in-events-and-actions" >}}) for how to retrieve context from vector stores leveraging built-in action and events. \ No newline at end of file +* See [Vector Stores]({{< ref "docs/development/vector_stores#built-in-events-and-actions" >}}) for how to retrieve context from vector stores leveraging built-in action and events. diff --git a/python/flink_agents/api/events/chat_event.py b/python/flink_agents/api/events/chat_event.py index a3a3deec1..0a4035ad7 100644 --- a/python/flink_agents/api/events/chat_event.py +++ b/python/flink_agents/api/events/chat_event.py @@ -83,8 +83,7 @@ def from_event(cls, event: Event) -> "ChatRequestEvent": prompt_args=event.attributes.get("prompt_args"), output_schema=output_schema_raw, ) - result.id = event.id - return result + return result.with_framework_metadata_from(event) @property def model(self) -> str: @@ -160,8 +159,7 @@ def from_event(cls, event: Event) -> "ChatResponseEvent": retry_count=event.attributes.get("retry_count", 0), total_retry_wait_sec=event.attributes.get("total_retry_wait_sec", 0), ) - result.id = event.id - return result + return result.with_framework_metadata_from(event) @property def request_id(self) -> UUID: diff --git a/python/flink_agents/api/events/context_retrieval_event.py b/python/flink_agents/api/events/context_retrieval_event.py index a8245a4ec..131aaf323 100644 --- a/python/flink_agents/api/events/context_retrieval_event.py +++ b/python/flink_agents/api/events/context_retrieval_event.py @@ -63,8 +63,7 @@ def from_event(cls, event: Event) -> "ContextRetrievalRequestEvent": vector_store=event.attributes["vector_store"], max_results=event.attributes.get("max_results", 3), ) - result.id = event.id - return result + return result.with_framework_metadata_from(event) @property def query(self) -> str: @@ -124,8 +123,7 @@ def from_event(cls, event: Event) -> "ContextRetrievalResponseEvent": query=event.attributes["query"], documents=documents, ) - result.id = event.id - return result + return result.with_framework_metadata_from(event) @property def request_id(self) -> UUID: diff --git a/python/flink_agents/api/events/event.py b/python/flink_agents/api/events/event.py index 77e7e5faf..01f4bc3d5 100644 --- a/python/flink_agents/api/events/event.py +++ b/python/flink_agents/api/events/event.py @@ -15,15 +15,14 @@ # See the License for the specific language governing permissions and # limitations under the License. ################################################################################# -import hashlib import json from typing import Any, ClassVar, Dict try: - from typing import override + from typing import Self, override except ImportError: - from typing_extensions import override -from uuid import UUID + from typing_extensions import Self, override +from uuid import UUID, uuid4 from pydantic import BaseModel, Field, model_validator from pydantic_core import PydanticSerializationError @@ -66,15 +65,14 @@ class Event(BaseModel, extra="allow"): Attributes: ---------- id : UUID - Unique identifier for the event, generated deterministically based on - event content. + Random version 4 UUID generated when the Event is created. type : str Event type string used for routing. Required for all events. attributes : Dict[str, Any] Key-value properties for the event data. """ - id: UUID = Field(default=None) + id: UUID = Field(default_factory=uuid4, frozen=True) type: str attributes: Dict[str, Any] = Field(default_factory=dict) @@ -98,23 +96,9 @@ def model_dump_json(self, **kwargs: Any) -> str: kwargs["fallback"] = self.__serialize_unknown return super().model_dump_json(**kwargs) - def _generate_content_based_id(self) -> UUID: - """Generate a deterministic UUID based on event content using MD5 hash. - - Similar to Java's UUID.nameUUIDFromBytes(), uses MD5 for version 3 UUID. - """ - # Serialize content excluding 'id' to avoid circular dependency - content_json = super().model_dump_json( - exclude={"id"}, fallback=self.__serialize_unknown - ) - md5_hash = hashlib.md5(content_json.encode()).digest() - return UUID(bytes=md5_hash, version=3) - @model_validator(mode="after") - def validate_and_set_id(self) -> "Event": - """Validate that fields are serializable and generate content-based ID.""" - if self.id is None: - object.__setattr__(self, "id", self._generate_content_based_id()) + def validate_serializable_fields(self) -> "Event": + """Validate that all Event fields can be serialized.""" self.model_dump_json() return self @@ -122,9 +106,10 @@ def __setattr__(self, name: str, value: Any) -> None: super().__setattr__(name, value) # Ensure added property can be serialized. self.model_dump_json() - # Regenerate ID if content changed (but not if setting 'id' itself) - if name != "id": - object.__setattr__(self, "id", self._generate_content_based_id()) + + def with_framework_metadata_from(self, source: "Event") -> Self: + """Return a copy with framework-owned metadata from the source Event.""" + return self.model_copy(update={"id": source.id}) def get_type(self) -> str: """Return the event type string used for routing.""" @@ -200,8 +185,7 @@ def __init__(self, input: Any) -> None: def from_event(cls, event: Event) -> "InputEvent": assert "input" in event.attributes result = InputEvent(input=event.attributes["input"]) - result.id = event.id - return result + return result.with_framework_metadata_from(event) @property def input(self) -> Any: @@ -233,8 +217,7 @@ def __init__(self, output: Any) -> None: def from_event(cls, event: Event) -> "OutputEvent": assert "output" in event.attributes result = OutputEvent(output=event.attributes["output"]) - result.id = event.id - return result + return result.with_framework_metadata_from(event) @property def output(self) -> Any: diff --git a/python/flink_agents/api/events/tool_event.py b/python/flink_agents/api/events/tool_event.py index b7396b6f7..71d46fc82 100644 --- a/python/flink_agents/api/events/tool_event.py +++ b/python/flink_agents/api/events/tool_event.py @@ -58,8 +58,7 @@ def from_event(cls, event: Event) -> "ToolRequestEvent": model=event.attributes["model"], tool_calls=event.attributes["tool_calls"], ) - result.id = event.id - return result + return result.with_framework_metadata_from(event) @property def model(self) -> str: @@ -120,13 +119,10 @@ def from_event(cls, event: Event) -> "ToolResponseEvent": request_id=event.attributes["request_id"], responses=responses, external_ids=event.attributes.get("external_ids", {}), - success=event.attributes.get( - "success", dict.fromkeys(responses, True) - ), + success=event.attributes.get("success", dict.fromkeys(responses, True)), error=event.attributes.get("error", {}), ) - result.id = event.id - return result + return result.with_framework_metadata_from(event) @property def request_id(self) -> UUID: diff --git a/python/flink_agents/api/tests/test_cross_language_event_snapshots.py b/python/flink_agents/api/tests/test_cross_language_event_snapshots.py index 315301aeb..83648967a 100644 --- a/python/flink_agents/api/tests/test_cross_language_event_snapshots.py +++ b/python/flink_agents/api/tests/test_cross_language_event_snapshots.py @@ -50,8 +50,7 @@ def _regenerate_enabled() -> bool: def _force_id(event: Event, fixed_id: UUID) -> Event: - object.__setattr__(event, "id", fixed_id) - return event + return event.model_copy(update={"id": fixed_id}) def _write_python_snapshot(name: str, event: Event) -> None: diff --git a/python/flink_agents/api/tests/test_event.py b/python/flink_agents/api/tests/test_event.py index 110106ea4..baf5e6244 100644 --- a/python/flink_agents/api/tests/test_event.py +++ b/python/flink_agents/api/tests/test_event.py @@ -16,7 +16,8 @@ # limitations under the License. ################################################################################# import json -from typing import Any, Type +from typing import Any, ClassVar, Type +from uuid import uuid4 import pytest from pydantic import ValidationError @@ -26,6 +27,18 @@ from flink_agents.api.events.event import Event, InputEvent, OutputEvent +class _CustomEvent(Event): + EVENT_TYPE: ClassVar[str] = "custom_event" + + def __init__(self, value: str) -> None: + super().__init__(type=self.EVENT_TYPE, attributes={"value": value}) + + @classmethod + def from_event(cls, event: Event) -> "_CustomEvent": + result = cls(value=event.attributes["value"]) + return result.with_framework_metadata_from(event) + + def test_event_init_serializable() -> None: Event(type="test", a=1, b=InputEvent(input=1), c=OutputEvent(output="111")) @@ -52,7 +65,11 @@ def test_input_event_ignore_row_unserializable() -> None: def test_event_row_with_non_serializable_fails() -> None: with pytest.raises(ValidationError): - Event(type="test", row_field=Row({"a": 1}), non_serializable_field=Type[InputEvent]) + Event( + type="test", + row_field=Row({"a": 1}), + non_serializable_field=Type[InputEvent], + ) def test_event_multiple_rows_serializable() -> None: @@ -64,6 +81,31 @@ def test_event_setattr_row_serializable() -> None: event.row_field = Row({"key": "value"}) +def test_events_with_same_content_have_distinct_random_ids() -> None: + first = OutputEvent(output="same") + second = OutputEvent(output="same") + + assert first.id != second.id + assert first.id.version == 4 + assert second.id.version == 4 + + +def test_event_id_does_not_change_with_event_content() -> None: + event = Event(type="test", a=1) + event_id = event.id + + event.a = 2 + + assert event.id == event_id + + +def test_event_id_cannot_be_reassigned() -> None: + event = Event(type="test") + + with pytest.raises(ValidationError, match="Field is frozen"): + event.id = uuid4() + + def test_event_json_serialization_with_row() -> None: event = InputEvent(input=Row({"test": "data"})) json_str = event.model_dump_json() @@ -203,6 +245,18 @@ def test_unified_event_serialization_roundtrip() -> None: assert restored.attributes == {"a": 1, "b": "two"} +def test_custom_typed_from_event_preserves_identity() -> None: + """Test custom typed reconstruction preserves framework-managed identity.""" + event = Event( + type=_CustomEvent.EVENT_TYPE, + attributes={"value": "result"}, + ) + + reconstructed = _CustomEvent.from_event(event) + + assert reconstructed.id == event.id + + def test_unified_event_serialization_roundtrip_with_row() -> None: """Test that unified events with Row fields survive JSON roundtrip.""" original = Event( diff --git a/python/flink_agents/e2e_tests/e2e_tests_integration/flink_integration_agent.py b/python/flink_agents/e2e_tests/e2e_tests_integration/flink_integration_agent.py index b464754f5..539868fc3 100644 --- a/python/flink_agents/e2e_tests/e2e_tests_integration/flink_integration_agent.py +++ b/python/flink_agents/e2e_tests/e2e_tests_integration/flink_integration_agent.py @@ -65,7 +65,8 @@ def __init__(self, value: Any) -> None: def from_event(cls, event: "Event") -> "MyEvent": """Reconstruct a MyEvent from a generic Event.""" assert "value" in event.attributes, "Missing 'value' in event attributes" - return cls(value=event.attributes["value"]) + result = cls(value=event.attributes["value"]) + return result.with_framework_metadata_from(event) @property def value(self) -> Any: diff --git a/python/flink_agents/e2e_tests/e2e_tests_integration/long_term_memory_test.py b/python/flink_agents/e2e_tests/e2e_tests_integration/long_term_memory_test.py index 3db2ae010..d0be81da9 100644 --- a/python/flink_agents/e2e_tests/e2e_tests_integration/long_term_memory_test.py +++ b/python/flink_agents/e2e_tests/e2e_tests_integration/long_term_memory_test.py @@ -120,7 +120,8 @@ def __init__(self, value: Any) -> None: def from_event(cls, event: Event) -> "MyEvent": """Reconstruct a MyEvent from a generic Event.""" assert "value" in event.attributes - return MyEvent(value=Record.model_validate(event.attributes["value"])) + result = cls(value=Record.model_validate(event.attributes["value"])) + return result.with_framework_metadata_from(event) @property def value(self) -> Any: diff --git a/python/flink_agents/runtime/tests/test_python_java_utils.py b/python/flink_agents/runtime/tests/test_python_java_utils.py index 7aeeafe9c..0d22b04e4 100644 --- a/python/flink_agents/runtime/tests/test_python_java_utils.py +++ b/python/flink_agents/runtime/tests/test_python_java_utils.py @@ -17,9 +17,15 @@ ################################################################################# import json +import cloudpickle + from flink_agents.api.decorators import tool +from flink_agents.api.events.event import Event from flink_agents.api.tools import InjectedArg -from flink_agents.runtime.python_java_utils import get_python_tool_metadata +from flink_agents.runtime.python_java_utils import ( + get_python_tool_metadata, + wrap_to_input_event, +) @tool(injected_args={"tenant_id": InjectedArg.from_config("tenant.id")}) @@ -36,6 +42,15 @@ def test_get_python_tool_metadata_merges_callable_injected_args() -> None: schema = json.loads(flat["inputSchema"]) assert set(schema["properties"]) == {"order_id"} injected_args = json.loads(flat["injectedArgs"]) - assert injected_args == { - "tenant_id": {"source": "config", "key": "tenant.id"} - } + assert injected_args == {"tenant_id": {"source": "config", "key": "tenant.id"}} + + +def test_wrap_to_input_event_assigns_unique_random_id_per_record() -> None: + payload = cloudpickle.dumps("same input") + + first = Event.from_json(wrap_to_input_event(payload)) + second = Event.from_json(wrap_to_input_event(payload)) + + assert first.id != second.id + assert first.id.version == 4 + assert second.id.version == 4 From affacb8a5393596ce8bb8000a103e79682a914c5 Mon Sep 17 00:00:00 2001 From: rosemaryYuan <91046107+rosemarYuan@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:50:25 +0800 Subject: [PATCH 2/3] [api][runtime][python] Add minimal event lineage --- .../org/apache/flink/agents/api/Event.java | 46 +++- .../apache/flink/agents/api/InputEvent.java | 4 +- .../apache/flink/agents/api/OutputEvent.java | 4 +- .../agents/api/event/ChatRequestEvent.java | 4 +- .../agents/api/event/ChatResponseEvent.java | 4 +- .../event/ContextRetrievalRequestEvent.java | 4 +- .../event/ContextRetrievalResponseEvent.java | 4 +- .../agents/api/event/ToolRequestEvent.java | 4 +- .../agents/api/event/ToolResponseEvent.java | 4 +- .../docs/development/workflow_agent.md | 15 +- docs/content/docs/operations/monitoring.md | 38 +++- python/flink_agents/api/events/event.py | 47 +++- .../runtime/eventlog/JsonTruncator.java | 24 ++- .../operator/ActionExecutionOperator.java | 11 +- tools/reconstruct_trace_tree.py | 203 ++++++++++++++++++ 15 files changed, 362 insertions(+), 54 deletions(-) create mode 100755 tools/reconstruct_trace_tree.py 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 e7fbde464..5b2e1480a 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 @@ -20,9 +20,12 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.databind.ObjectMapper; +import javax.annotation.Nullable; + import java.io.IOException; import java.util.HashMap; import java.util.Map; @@ -38,6 +41,9 @@ public class Event { private final String type; private final Map attributes; + @Nullable private UUID upstreamEventId; + @Nullable private String upstreamActionName; + /** * Runtime-internal timestamp from the source record. Not part of the cross-language event * contract; used by the Flink runtime for timestamp propagation. @@ -81,6 +87,30 @@ public Map getAttributes() { return attributes; } + /** Returns the ID of the Event consumed by the Action that emitted this Event. */ + @Nullable + @JsonInclude(JsonInclude.Include.NON_NULL) + public UUID getUpstreamEventId() { + return upstreamEventId; + } + + /** Sets the ID of the Event consumed by the Action that emitted this Event. */ + public void setUpstreamEventId(@Nullable UUID upstreamEventId) { + this.upstreamEventId = upstreamEventId; + } + + /** Returns the name of the Action that emitted this Event. */ + @Nullable + @JsonInclude(JsonInclude.Include.NON_NULL) + public String getUpstreamActionName() { + return upstreamActionName; + } + + /** Sets the name of the Action that emitted this Event. */ + public void setUpstreamActionName(@Nullable String upstreamActionName) { + this.upstreamActionName = upstreamActionName; + } + public Object getAttr(String name) { return attributes.get(name); } @@ -105,18 +135,24 @@ public void setSourceTimestamp(long timestamp) { } /** - * Creates a base Event from another Event, copying id, type, and attributes. Subclasses - * override this to reconstruct typed event objects with proper field deserialization. + * Creates a base Event from another Event, copying its identity, data, and framework metadata. + * Subclasses override this to reconstruct typed event objects with proper field + * deserialization. */ public static Event fromEvent(Event event) { Event copy = new Event(event.getId(), event.getType(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - copy.setSourceTimestamp(event.getSourceTimestamp()); - } + copy.copyFrameworkMetadataFrom(event); return copy; } + /** Copies framework-managed metadata when reconstructing a typed Event. */ + protected void copyFrameworkMetadataFrom(Event event) { + this.sourceTimestamp = event.sourceTimestamp; + this.upstreamEventId = event.upstreamEventId; + this.upstreamActionName = event.upstreamActionName; + } + /** * Creates an Event from a JSON string. * diff --git a/api/src/main/java/org/apache/flink/agents/api/InputEvent.java b/api/src/main/java/org/apache/flink/agents/api/InputEvent.java index d07370215..c3328c03a 100644 --- a/api/src/main/java/org/apache/flink/agents/api/InputEvent.java +++ b/api/src/main/java/org/apache/flink/agents/api/InputEvent.java @@ -51,9 +51,7 @@ public InputEvent( */ public static InputEvent fromEvent(Event event) { InputEvent result = new InputEvent(event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } diff --git a/api/src/main/java/org/apache/flink/agents/api/OutputEvent.java b/api/src/main/java/org/apache/flink/agents/api/OutputEvent.java index d7a7e0f4a..b310b1cb1 100644 --- a/api/src/main/java/org/apache/flink/agents/api/OutputEvent.java +++ b/api/src/main/java/org/apache/flink/agents/api/OutputEvent.java @@ -54,9 +54,7 @@ public OutputEvent( */ public static OutputEvent fromEvent(Event event) { OutputEvent result = new OutputEvent(event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } diff --git a/api/src/main/java/org/apache/flink/agents/api/event/ChatRequestEvent.java b/api/src/main/java/org/apache/flink/agents/api/event/ChatRequestEvent.java index 4e05d1c88..7c2c7d03b 100644 --- a/api/src/main/java/org/apache/flink/agents/api/event/ChatRequestEvent.java +++ b/api/src/main/java/org/apache/flink/agents/api/event/ChatRequestEvent.java @@ -98,9 +98,7 @@ private static Map normalizeAttributes(Map attri public static ChatRequestEvent fromEvent(Event event) { ChatRequestEvent result = new ChatRequestEvent(event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } diff --git a/api/src/main/java/org/apache/flink/agents/api/event/ChatResponseEvent.java b/api/src/main/java/org/apache/flink/agents/api/event/ChatResponseEvent.java index 8dab3b1d8..20ce8d9ca 100644 --- a/api/src/main/java/org/apache/flink/agents/api/event/ChatResponseEvent.java +++ b/api/src/main/java/org/apache/flink/agents/api/event/ChatResponseEvent.java @@ -77,9 +77,7 @@ private static Map normalizeAttributes(Map attri public static ChatResponseEvent fromEvent(Event event) { ChatResponseEvent result = new ChatResponseEvent(event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } diff --git a/api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalRequestEvent.java b/api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalRequestEvent.java index 387319c95..625a27420 100644 --- a/api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalRequestEvent.java +++ b/api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalRequestEvent.java @@ -62,9 +62,7 @@ public static ContextRetrievalRequestEvent fromEvent(Event event) { ContextRetrievalRequestEvent result = new ContextRetrievalRequestEvent( event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } diff --git a/api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalResponseEvent.java b/api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalResponseEvent.java index a71fd7031..2635bfc93 100644 --- a/api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalResponseEvent.java +++ b/api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalResponseEvent.java @@ -85,9 +85,7 @@ public static ContextRetrievalResponseEvent fromEvent(Event event) { ContextRetrievalResponseEvent result = new ContextRetrievalResponseEvent( event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } diff --git a/api/src/main/java/org/apache/flink/agents/api/event/ToolRequestEvent.java b/api/src/main/java/org/apache/flink/agents/api/event/ToolRequestEvent.java index 9a24ef7ea..db7567cd7 100644 --- a/api/src/main/java/org/apache/flink/agents/api/event/ToolRequestEvent.java +++ b/api/src/main/java/org/apache/flink/agents/api/event/ToolRequestEvent.java @@ -56,9 +56,7 @@ public ToolRequestEvent( public static ToolRequestEvent fromEvent(Event event) { ToolRequestEvent result = new ToolRequestEvent(event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } diff --git a/api/src/main/java/org/apache/flink/agents/api/event/ToolResponseEvent.java b/api/src/main/java/org/apache/flink/agents/api/event/ToolResponseEvent.java index 896f3bcb7..10e6e849b 100644 --- a/api/src/main/java/org/apache/flink/agents/api/event/ToolResponseEvent.java +++ b/api/src/main/java/org/apache/flink/agents/api/event/ToolResponseEvent.java @@ -100,9 +100,7 @@ private static Map normalizeAttributes(Map attri public static ToolResponseEvent fromEvent(Event event) { ToolResponseEvent result = new ToolResponseEvent(event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } diff --git a/docs/content/docs/development/workflow_agent.md b/docs/content/docs/development/workflow_agent.md index 0540c239d..46e84cdd9 100644 --- a/docs/content/docs/development/workflow_agent.md +++ b/docs/content/docs/development/workflow_agent.md @@ -685,12 +685,8 @@ public class MyEvent extends Event { } public static MyEvent fromEvent(Event event) { - // Preserve the base event id (and sourceTimestamp) so event logs, listeners, - // correlation, deduplication, and timestamp propagation stay consistent. MyEvent result = new MyEvent(event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } @@ -708,10 +704,11 @@ Typed reconstruction represents the same Event occurrence, so it must preserve t identity and framework-managed metadata: - **Python**: return `with_framework_metadata_from(event)` after constructing the typed object. It - returns a new typed object with the Event's UUIDv4 `id`; the returned Event's `id` remains - immutable. -- **Java**: pass `event.getId()` to the typed constructor, then carry over `sourceTimestamp` with - `setSourceTimestamp(...)` when `hasSourceTimestamp()` is true. + returns a new typed object with the Event's UUIDv4 `id`, `upstream_event_id`, and + `upstream_action_name`; the returned Event's `id` remains immutable. +- **Java**: pass `event.getId()` to the typed constructor, then call + `copyFrameworkMetadataFrom(event)`. The helper preserves `sourceTimestamp`, + `upstreamEventId`, and `upstreamActionName`. {{< /hint >}} {{< hint info >}} diff --git a/docs/content/docs/operations/monitoring.md b/docs/content/docs/operations/monitoring.md index 021452125..4df0963e6 100644 --- a/docs/content/docs/operations/monitoring.md +++ b/docs/content/docs/operations/monitoring.md @@ -184,14 +184,44 @@ Each record contains a top-level `timestamp`, the resolved `logLevel`, and a top { "timestamp": "2024-01-15T10:30:00Z", "logLevel": "STANDARD", - "eventType": "_input_event", + "eventType": "MiddleEvent", "event": { - "eventType": "_input_event", - "...": "..." + "eventType": "MiddleEvent", + "id": "39361629-4f1d-4b62-b734-a48b181cb6e0", + "attributes": {}, + "type": "MiddleEvent", + "upstreamEventId": "dad5ed00-80e2-4746-8bdb-f126bad504b5", + "upstreamActionName": "action1" } } ``` +`upstreamEventId` identifies the Event consumed by the Action that emitted the current Event, and `upstreamActionName` identifies that Action. The framework maintains both fields directly on the `event` object, outside business `attributes`. A root `InputEvent` omits both fields. + +### Trace Tree Reconstruction + +The local reader rebuilds InputEvent-rooted Trace Trees from a saved File Event Log. From the repository root, pass either one log file for text output or a log directory for Trace Tree JSON: + +```bash +python tools/reconstruct_trace_tree.py /path/to/events-job-task-0.log +python tools/reconstruct_trace_tree.py /path/to/event-log-directory --format json +``` + +For example, the text output for an `InputEvent -> MiddleEvent -> OutputEvent` lineage is: + +```text +Trace Tree 1 + _input_event (dad5ed00-80e2-4746-8bdb-f126bad504b5) + [Action: action1] + MiddleEvent (39361629-4f1d-4b62-b734-a48b181cb6e0) + [Action: action2] + _output_event (f452ce08-c672-4c9c-841f-d81d15a900c5) +``` + +In JSON output, `roots` lists root Event IDs and `nodes` maps each Event ID to its Event node. The reader derives virtual Action nodes from `upstreamActionName`; they are not separate Event Log records. Log order controls display order only and does not add execution-order semantics. + +Reconstruction warnings are written to standard error and included in JSON output while valid InputEvent-rooted trees are retained. Warnings cover duplicate Event IDs, missing parent Events, missing Action names, non-InputEvents without parents, and InputEvents with invalid upstream lineage. + ### Event Log Levels Each event type is logged at a configurable verbosity. Three levels are supported: @@ -204,7 +234,7 @@ Each event type is logged at a configurable verbosity. Three levels are supporte The global default is set by [`event-log.level`]({{< ref "docs/operations/configuration#core-options" >}}). At `STANDARD` level, the payload is shrunk along three independent axes — long strings, large arrays, and deep nesting — controlled by `event-log.standard.max-string-length`, `event-log.standard.max-array-elements`, and `event-log.standard.max-depth` respectively. Setting any threshold to `0` disables that specific truncation; setting all three to `0` makes `STANDARD` behave identically to `VERBOSE` (apart from the `logLevel` label). The exact truncation strategy may evolve over time; the contract is only that `STANDARD` keeps logs concise while `VERBOSE` preserves the full payload. -**Fields that are never truncated.** Structural and identifying fields are always preserved in full so log consumers can still group, route, and correlate records: `timestamp`, `logLevel`, top-level `eventType`, and the event's own `id`, `type`, and short scalar fields. Truncation only applies to large nested content (long strings, big arrays, deeply nested objects). +**Fields that are never truncated.** Structural and identifying fields are always preserved in full so log consumers can still group, route, and correlate records: `timestamp`, `logLevel`, top-level `eventType`, and the event's own `eventType`, `type`, `id`, `upstreamEventId`, and `upstreamActionName`. The `attributes` envelope is also preserved, while large nested content inside it can still be truncated. **Truncation wrapper format.** When a field is truncated at `STANDARD` level it is replaced by a JSON object that records what was retained and what was dropped. This keeps the record valid JSON and lets downstream tooling detect truncation programmatically: diff --git a/python/flink_agents/api/events/event.py b/python/flink_agents/api/events/event.py index 01f4bc3d5..0d4f34884 100644 --- a/python/flink_agents/api/events/event.py +++ b/python/flink_agents/api/events/event.py @@ -24,7 +24,14 @@ from typing_extensions import Self, override from uuid import UUID, uuid4 -from pydantic import BaseModel, Field, model_validator +from pydantic import ( + AliasChoices, + BaseModel, + Field, + SerializerFunctionWrapHandler, + model_serializer, + model_validator, +) from pydantic_core import PydanticSerializationError from pyflink.common import Row @@ -70,11 +77,25 @@ class Event(BaseModel, extra="allow"): Event type string used for routing. Required for all events. attributes : Dict[str, Any] Key-value properties for the event data. + upstream_event_id : UUID | None + ID of the direct Event consumed by the Action that emitted this Event. + upstream_action_name : str | None + Name of the Action that emitted this Event. """ id: UUID = Field(default_factory=uuid4, frozen=True) type: str attributes: Dict[str, Any] = Field(default_factory=dict) + upstream_event_id: UUID | None = Field( + default=None, + validation_alias=AliasChoices("upstream_event_id", "upstreamEventId"), + serialization_alias="upstreamEventId", + ) + upstream_action_name: str | None = Field( + default=None, + validation_alias=AliasChoices("upstream_action_name", "upstreamActionName"), + serialization_alias="upstreamActionName", + ) @staticmethod def __serialize_unknown(field: Any) -> Dict[str, Any]: @@ -94,8 +115,24 @@ def model_dump_json(self, **kwargs: Any) -> str: # Set fallback if not provided in kwargs if "fallback" not in kwargs: kwargs["fallback"] = self.__serialize_unknown + if "by_alias" not in kwargs: + kwargs["by_alias"] = True return super().model_dump_json(**kwargs) + @model_serializer(mode="wrap") + def _serialize_event( + self, handler: SerializerFunctionWrapHandler + ) -> Dict[str, Any]: + """Omit empty lineage without changing other null-valued business data.""" + serialized: Dict[str, Any] = handler(self) + if self.upstream_event_id is None: + serialized.pop("upstream_event_id", None) + serialized.pop("upstreamEventId", None) + if self.upstream_action_name is None: + serialized.pop("upstream_action_name", None) + serialized.pop("upstreamActionName", None) + return serialized + @model_validator(mode="after") def validate_serializable_fields(self) -> "Event": """Validate that all Event fields can be serialized.""" @@ -109,7 +146,13 @@ def __setattr__(self, name: str, value: Any) -> None: def with_framework_metadata_from(self, source: "Event") -> Self: """Return a copy with framework-owned metadata from the source Event.""" - return self.model_copy(update={"id": source.id}) + return self.model_copy( + update={ + "id": source.id, + "upstream_event_id": source.upstream_event_id, + "upstream_action_name": source.upstream_action_name, + } + ) def get_type(self) -> str: """Return the event type string used for routing.""" diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/JsonTruncator.java b/runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/JsonTruncator.java index 89e49ceeb..fa89a95aa 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/JsonTruncator.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/JsonTruncator.java @@ -24,8 +24,6 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import java.util.ArrayList; -import java.util.Arrays; -import java.util.HashSet; import java.util.List; import java.util.Set; @@ -46,14 +44,21 @@ *

Setting any threshold to {@code 0} disables that specific truncation strategy. If all * thresholds are {@code 0}, no truncation occurs. * - *

Protected fields at the top level of the event node ({@code eventType}, {@code id}, {@code - * attributes}) are never truncated as structural fields. The {@code attributes} envelope is - * additionally traversed so user payload stored inside it is truncated like any other field. + *

Protected fields at the top level of the event node ({@code eventType}, {@code type}, {@code + * id}, {@code upstreamEventId}, {@code upstreamActionName}, {@code attributes}) are never truncated + * as structural fields. The {@code attributes} envelope is additionally traversed so user payload + * stored inside it is truncated like any other field. */ public class JsonTruncator { private static final Set PROTECTED_FIELDS = - new HashSet<>(Arrays.asList("eventType", "id", "attributes")); + Set.of( + "eventType", + "type", + "id", + "upstreamEventId", + "upstreamActionName", + "attributes"); private final int maxStringLength; private final int maxArrayElements; @@ -75,9 +80,10 @@ public JsonTruncator(int maxStringLength, int maxArrayElements, int maxDepth) { /** * Truncates the given event node in place according to configured thresholds. * - *

Protected fields ({@code eventType}, {@code id}, {@code attributes}) at the top level of - * the event node are never truncated as structural fields. The {@code attributes} envelope is - * additionally traversed so user payload stored inside it is truncated like any other field. + *

Protected fields ({@code eventType}, {@code type}, {@code id}, {@code upstreamEventId}, + * {@code upstreamActionName}, {@code attributes}) at the top level of the event node are never + * truncated as structural fields. The {@code attributes} envelope is additionally traversed so + * user payload stored inside it is truncated like any other field. * * @param eventNode the top-level event JSON object to truncate * @return {@code true} if any field was truncated, {@code false} if the node was unchanged 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 46d09e0a2..61dae4c50 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 @@ -346,6 +346,7 @@ private void processActionTaskForKey(Object key) throws Exception { key); isFinished = true; outputEvents = actionState.getOutputEvents(); + setOutputEventLineage(actionTask, outputEvents); for (MemoryUpdate memoryUpdate : actionState.getShortTermMemoryUpdates()) { actionTask .getRunnerContext() @@ -385,6 +386,8 @@ private void processActionTaskForKey(Object key) throws Exception { durableExecManager.removeDurableContext(actionTask); contextManager.removeContinuationContext(actionTask); contextManager.removePythonAwaitableRef(actionTask); + outputEvents = actionTaskResult.getOutputEvents(); + setOutputEventLineage(actionTask, outputEvents); durableExecManager.maybePersistTaskResult( key, sequenceNumber, @@ -393,7 +396,6 @@ private void processActionTaskForKey(Object key) throws Exception { actionTask.getRunnerContext(), actionTaskResult); isFinished = actionTaskResult.isFinished(); - outputEvents = actionTaskResult.getOutputEvents(); generatedActionTaskOpt = actionTaskResult.getGeneratedActionTask(); } @@ -454,6 +456,13 @@ private void processActionTaskForKey(Object key) throws Exception { } } + private static void setOutputEventLineage(ActionTask actionTask, List outputEvents) { + for (Event outputEvent : outputEvents) { + outputEvent.setUpstreamEventId(actionTask.event.getId()); + outputEvent.setUpstreamActionName(actionTask.action.getName()); + } + } + @Override public void endInput() throws Exception { waitInFlightEventsFinished(); diff --git a/tools/reconstruct_trace_tree.py b/tools/reconstruct_trace_tree.py new file mode 100755 index 000000000..d7eab1d2a --- /dev/null +++ b/tools/reconstruct_trace_tree.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +# 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. +"""Reconstruct InputEvent-rooted Trace Trees from an Event Log.""" + +import argparse +import json +import sys +from collections import defaultdict +from pathlib import Path +from typing import Any, Iterator + +INPUT_EVENT_TYPE = "_input_event" + + +def find_log_files(path: Path) -> list[Path]: + """Return Event Log files in deterministic file-name order.""" + if path.is_file(): + return [path] + log_files = sorted(path.glob("events-*.log")) + return log_files or sorted(path.glob("*.log")) + + +def read_json_objects(path: Path) -> Iterator[dict[str, Any]]: + """Read consecutive JSON objects separated by whitespace.""" + content = path.read_text(encoding="utf-8") + decoder = json.JSONDecoder() + position = 0 + while position < len(content): + while position < len(content) and content[position].isspace(): + position += 1 + if position == len(content): + return + record, position = decoder.raw_decode(content, position) + yield record + + +def read_event_records(path: Path) -> Iterator[dict[str, Any]]: + """Read the fields needed to reconstruct Trace Trees.""" + log_files = find_log_files(path) + if not log_files: + message = f"No Event Log files found at {path}" + raise FileNotFoundError(message) + + for log_file in log_files: + for record in read_json_objects(log_file): + event = record["event"] + yield { + "eventId": str(event["id"]), + "eventType": record["eventType"], + "timestamp": record.get("timestamp"), + "upstreamEventId": event.get("upstreamEventId"), + "upstreamActionName": event.get("upstreamActionName"), + } + + +def build_trace_forest(records: Iterator[dict[str, Any]]) -> dict[str, Any]: + """Build valid InputEvent trees while retaining auditable invalid nodes.""" + records_by_id: dict[str, list[dict[str, Any]]] = defaultdict(list) + for record in records: + records_by_id[record["eventId"]].append(record) + + warnings: list[dict[str, str]] = [] + nodes: dict[str, dict[str, Any]] = {} + for event_id, matching_records in records_by_id.items(): + if len(matching_records) > 1: + warnings.append( + warning( + "DUPLICATE_EVENT_ID", + event_id, + f"Event ID {event_id} appears {len(matching_records)} times.", + ) + ) + continue + record = matching_records[0] + nodes[event_id] = { + "eventId": event_id, + "eventType": record["eventType"], + "timestamp": record["timestamp"], + "upstreamEventId": record["upstreamEventId"], + "upstreamActionName": record["upstreamActionName"], + "actions": [], + } + + roots: list[str] = [] + action_indexes: dict[str, dict[str, dict[str, Any]]] = defaultdict(dict) + for event_id, node in nodes.items(): + upstream_event_id = node["upstreamEventId"] + upstream_action_name = node["upstreamActionName"] + + if node["eventType"] == INPUT_EVENT_TYPE: + if upstream_event_id is None and upstream_action_name is None: + roots.append(event_id) + else: + warnings.append( + warning( + "INVALID_ROOT_LINEAGE", + event_id, + f"InputEvent {event_id} must not have upstream lineage.", + ) + ) + continue + + if upstream_event_id is None: + warnings.append( + warning( + "UNLINKED_EVENT", + event_id, + f"Non-InputEvent {event_id} has no upstream Event.", + ) + ) + continue + if upstream_action_name is None: + warnings.append( + warning( + "MISSING_ACTION_NAME", + event_id, + f"Event {event_id} has no upstream Action name.", + ) + ) + continue + if upstream_event_id not in nodes: + warnings.append( + warning( + "MISSING_PARENT", + event_id, + f"Event {event_id} references missing parent {upstream_event_id}.", + ) + ) + continue + + action = action_indexes[upstream_event_id].get(upstream_action_name) + if action is None: + action = {"name": upstream_action_name, "children": []} + action_indexes[upstream_event_id][upstream_action_name] = action + nodes[upstream_event_id]["actions"].append(action) + action["children"].append(event_id) + + return {"roots": roots, "nodes": nodes, "warnings": warnings} + + +def warning(code: str, event_id: str, message: str) -> dict[str, str]: + """Create one machine-readable reconstruction warning.""" + return {"code": code, "eventId": event_id, "message": message} + + +def render_text(trace_forest: dict[str, Any]) -> str: + """Render valid Trace Trees without assigning meaning to sibling order.""" + lines: list[str] = [] + + def render_event(event_id: str, indent: str) -> None: + stack: list[tuple[str, Any, str]] = [("event", event_id, indent)] + while stack: + item_type, item, item_indent = stack.pop() + if item_type == "event": + node = trace_forest["nodes"][item] + lines.append(f"{item_indent}{node['eventType']} ({item})") + for action in reversed(node["actions"]): + stack.append(("action", action, item_indent)) + else: + lines.append(f"{item_indent} [Action: {item['name']}]") + for child_id in reversed(item["children"]): + stack.append(("event", child_id, item_indent + " ")) + + for root_number, root_id in enumerate(trace_forest["roots"], start=1): + if lines: + lines.append("") + lines.append(f"Trace Tree {root_number}") + render_event(root_id, " ") + return "\n".join(lines) + + +def main() -> None: + """Run the command-line reader.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("path", type=Path, help="Event Log file or directory") + parser.add_argument("--format", choices=("text", "json"), default="text") + args = parser.parse_args() + + trace_forest = build_trace_forest(read_event_records(args.path)) + for item in trace_forest["warnings"]: + print(f"[{item['code']}] {item['message']}", file=sys.stderr) + + if args.format == "json": + print(json.dumps(trace_forest, indent=2)) + else: + print(render_text(trace_forest)) + + +if __name__ == "__main__": + main() From de60cc5a975104c706f5892d2d7f988830793707 Mon Sep 17 00:00:00 2001 From: rosemaryYuan <91046107+rosemarYuan@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:50:26 +0800 Subject: [PATCH 3/3] [test][java][python] Verify minimal event lineage --- .../apache/flink/agents/api/EventTest.java | 57 ++++++ python/flink_agents/api/tests/test_event.py | 56 +++++- .../api/tests/test_trace_tree_reader.py | 186 ++++++++++++++++++ .../python_event_logging_test.py | 80 ++++++++ .../eventlog/EventLogRecordJsonSerdeTest.java | 29 ++- .../runtime/eventlog/FileEventLoggerTest.java | 10 +- .../runtime/eventlog/JsonTruncatorTest.java | 13 +- .../operator/ActionExecutionOperatorTest.java | 76 +++++++ .../EventLineageReconstructionTest.java | 127 ++++++++++++ 9 files changed, 623 insertions(+), 11 deletions(-) create mode 100644 python/flink_agents/api/tests/test_trace_tree_reader.py create mode 100644 runtime/src/test/java/org/apache/flink/agents/runtime/operator/EventLineageReconstructionTest.java diff --git a/api/src/test/java/org/apache/flink/agents/api/EventTest.java b/api/src/test/java/org/apache/flink/agents/api/EventTest.java index c4a4ca104..b3ab4207d 100644 --- a/api/src/test/java/org/apache/flink/agents/api/EventTest.java +++ b/api/src/test/java/org/apache/flink/agents/api/EventTest.java @@ -143,6 +143,31 @@ void testSubclassedEventJsonRoundTrip() throws Exception { assertEquals(InputEvent.EVENT_TYPE, deserialized.getType()); } + @Test + void testLineageJsonRoundTrip() throws Exception { + UUID upstreamEventId = UUID.randomUUID(); + Event original = new Event("ChildEvent"); + original.setUpstreamEventId(upstreamEventId); + original.setUpstreamActionName("child_action"); + + String json = objectMapper.writeValueAsString(original); + JsonNode node = objectMapper.readTree(json); + Event deserialized = objectMapper.readValue(json, Event.class); + + assertEquals(upstreamEventId.toString(), node.get("upstreamEventId").asText()); + assertEquals("child_action", node.get("upstreamActionName").asText()); + assertEquals(upstreamEventId, deserialized.getUpstreamEventId()); + assertEquals("child_action", deserialized.getUpstreamActionName()); + } + + @Test + void testRootEventOmitsLineageFieldsFromJson() throws Exception { + JsonNode node = objectMapper.readTree(objectMapper.writeValueAsString(new InputEvent(1L))); + + assertFalse(node.has("upstreamEventId")); + assertFalse(node.has("upstreamActionName")); + } + // ── fromJson ─────────────────────────────────────────────────────────── @Test @@ -218,4 +243,36 @@ void testSourceTimestamp() { assertTrue(event.hasSourceTimestamp()); assertEquals(123456789L, event.getSourceTimestamp()); } + + @Test + void testFromEventCopiesFrameworkMetadata() { + UUID upstreamEventId = UUID.randomUUID(); + Event original = new Event("Test"); + original.setSourceTimestamp(123456789L); + original.setUpstreamEventId(upstreamEventId); + original.setUpstreamActionName("test_action"); + + Event copy = Event.fromEvent(original); + + assertEquals(123456789L, copy.getSourceTimestamp()); + assertEquals(upstreamEventId, copy.getUpstreamEventId()); + assertEquals("test_action", copy.getUpstreamActionName()); + } + + @Test + void testTypedFromEventCopiesLineage() { + UUID upstreamEventId = UUID.randomUUID(); + Event original = + new Event( + UUID.randomUUID(), + OutputEvent.EVENT_TYPE, + new HashMap<>(Map.of("output", "result"))); + original.setUpstreamEventId(upstreamEventId); + original.setUpstreamActionName("output_action"); + + OutputEvent copy = OutputEvent.fromEvent(original); + + assertEquals(upstreamEventId, copy.getUpstreamEventId()); + assertEquals("output_action", copy.getUpstreamActionName()); + } } diff --git a/python/flink_agents/api/tests/test_event.py b/python/flink_agents/api/tests/test_event.py index baf5e6244..44fefd705 100644 --- a/python/flink_agents/api/tests/test_event.py +++ b/python/flink_agents/api/tests/test_event.py @@ -17,7 +17,7 @@ ################################################################################# import json from typing import Any, ClassVar, Type -from uuid import uuid4 +from uuid import UUID, uuid4 import pytest from pydantic import ValidationError @@ -245,16 +245,66 @@ def test_unified_event_serialization_roundtrip() -> None: assert restored.attributes == {"a": 1, "b": "two"} -def test_custom_typed_from_event_preserves_identity() -> None: - """Test custom typed reconstruction preserves framework-managed identity.""" +def test_event_lineage_json_roundtrip_uses_java_field_names() -> None: + """Test framework-managed lineage has a stable cross-language JSON shape.""" + upstream_event_id = UUID("00000000-0000-0000-0000-000000000001") + event = Event(type="ChildEvent") + event_id = event.id + + event.upstream_event_id = upstream_event_id + event.upstream_action_name = "child_action" + + parsed = json.loads(event.model_dump_json()) + restored = Event.from_json(json.dumps(parsed)) + + assert event.id == event_id + assert parsed["upstreamEventId"] == str(upstream_event_id) + assert parsed["upstreamActionName"] == "child_action" + assert "upstream_event_id" not in parsed + assert "upstream_action_name" not in parsed + assert restored.upstream_event_id == upstream_event_id + assert restored.upstream_action_name == "child_action" + + +def test_root_event_omits_lineage_fields_from_json() -> None: + """Test a root Event does not serialize empty lineage fields.""" + parsed = json.loads(InputEvent(input="root").model_dump_json()) + + assert "upstreamEventId" not in parsed + assert "upstreamActionName" not in parsed + + +def test_typed_from_event_preserves_lineage() -> None: + """Test typed reconstruction keeps framework-managed lineage metadata.""" + upstream_event_id = UUID("00000000-0000-0000-0000-000000000001") + event = Event( + type="_output_event", + attributes={"output": "result"}, + upstreamEventId=upstream_event_id, + upstreamActionName="output_action", + ) + + reconstructed = OutputEvent.from_event(event) + + assert reconstructed.upstream_event_id == upstream_event_id + assert reconstructed.upstream_action_name == "output_action" + + +def test_custom_typed_from_event_preserves_identity_and_lineage() -> None: + """Test custom typed reconstruction follows the framework metadata contract.""" + upstream_event_id = UUID("00000000-0000-0000-0000-000000000001") event = Event( type=_CustomEvent.EVENT_TYPE, attributes={"value": "result"}, + upstreamEventId=upstream_event_id, + upstreamActionName="custom_action", ) reconstructed = _CustomEvent.from_event(event) assert reconstructed.id == event.id + assert reconstructed.upstream_event_id == upstream_event_id + assert reconstructed.upstream_action_name == "custom_action" def test_unified_event_serialization_roundtrip_with_row() -> None: diff --git a/python/flink_agents/api/tests/test_trace_tree_reader.py b/python/flink_agents/api/tests/test_trace_tree_reader.py new file mode 100644 index 000000000..aa955b8ed --- /dev/null +++ b/python/flink_agents/api/tests/test_trace_tree_reader.py @@ -0,0 +1,186 @@ +################################################################################ +# 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. +################################################################################# +import json +import subprocess +import sys +from pathlib import Path + +_REPO_ROOT = Path(__file__).resolve().parents[4] +_READER = _REPO_ROOT / "tools" / "reconstruct_trace_tree.py" + + +def _record( + event_id: str, + event_type: str, + upstream_event_id: str | None = None, + upstream_action_name: str | None = None, +) -> dict: + event = {"id": event_id, "eventType": event_type, "attributes": {}} + if upstream_event_id is not None: + event["upstreamEventId"] = upstream_event_id + if upstream_action_name is not None: + event["upstreamActionName"] = upstream_action_name + return { + "timestamp": "2026-07-17T10:00:00Z", + "eventType": event_type, + "event": event, + } + + +def _write_log(path: Path, records: list[dict]) -> None: + path.write_text("".join(json.dumps(record) + "\n" for record in records)) + + +def _write_pretty_log(path: Path, records: list[dict]) -> None: + path.write_text("".join(json.dumps(record, indent=2) + "\n" for record in records)) + + +def _run_reader(log_path: Path, output_format: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, str(_READER), str(log_path), "--format", output_format], + check=True, + capture_output=True, + text=True, + ) + + +def test_reader_reconstructs_text_and_json_in_log_order(tmp_path: Path) -> None: + log_path = tmp_path / "events.log" + _write_log( + log_path, + [ + _record("root", "_input_event"), + _record("child-1", "ChildEvent", "root", "branch_action"), + _record("child-2", "ChildEvent", "root", "branch_action"), + _record("output", "_output_event", "child-1", "output_action"), + _record("child-3", "ChildEvent", "root", "second_action"), + ], + ) + + json_result = _run_reader(log_path, "json") + trace_forest = json.loads(json_result.stdout) + text_result = _run_reader(log_path, "text") + + assert trace_forest["roots"] == ["root"] + assert trace_forest["nodes"]["root"]["actions"] == [ + {"name": "branch_action", "children": ["child-1", "child-2"]}, + {"name": "second_action", "children": ["child-3"]}, + ] + assert trace_forest["nodes"]["child-1"]["actions"] == [ + {"name": "output_action", "children": ["output"]} + ] + assert trace_forest["warnings"] == [] + assert json_result.stderr == "" + expected_text_order = [ + "_input_event (root)", + "[Action: branch_action]", + "ChildEvent (child-1)", + "[Action: output_action]", + "_output_event (output)", + "ChildEvent (child-2)", + "[Action: second_action]", + "ChildEvent (child-3)", + ] + positions = [text_result.stdout.index(item) for item in expected_text_order] + assert positions == sorted(positions) + assert text_result.stderr == "" + + +def test_reader_reconstructs_pretty_printed_records(tmp_path: Path) -> None: + log_path = tmp_path / "events.log" + _write_pretty_log( + log_path, + [ + _record("root", "_input_event"), + _record("child-1", "ChildEvent", "root", "branch_action"), + _record("child-2", "ChildEvent", "root", "branch_action"), + ], + ) + + json_result = _run_reader(log_path, "json") + trace_forest = json.loads(json_result.stdout) + text_result = _run_reader(log_path, "text") + + assert trace_forest["roots"] == ["root"] + assert trace_forest["nodes"]["root"]["actions"] == [ + {"name": "branch_action", "children": ["child-1", "child-2"]} + ] + assert trace_forest["warnings"] == [] + assert "_input_event (root)" in text_result.stdout + assert "[Action: branch_action]" in text_result.stdout + assert text_result.stdout.index("ChildEvent (child-1)") < text_result.stdout.index( + "ChildEvent (child-2)" + ) + assert json_result.stderr == "" + assert text_result.stderr == "" + + +def test_text_reader_handles_trace_deeper_than_recursion_limit(tmp_path: Path) -> None: + log_path = tmp_path / "events.log" + depth = sys.getrecursionlimit() + 10 + records = [_record("event-0", "_input_event")] + records.extend( + _record( + f"event-{index}", + "MiddleEvent", + f"event-{index - 1}", + f"action-{index}", + ) + for index in range(1, depth + 1) + ) + _write_log(log_path, records) + + result = _run_reader(log_path, "text") + + assert f"MiddleEvent (event-{depth})" in result.stdout + assert result.stderr == "" + + +def test_reader_warns_and_keeps_valid_input_tree(tmp_path: Path) -> None: + log_path = tmp_path / "events.log" + _write_log( + log_path, + [ + _record("root", "_input_event"), + _record("valid-child", "ChildEvent", "root", "valid_action"), + _record("unlinked", "UnlinkedEvent"), + _record("missing", "MissingEvent", "absent", "missing_action"), + _record("duplicate", "FirstDuplicate"), + _record("duplicate", "SecondDuplicate", "root", "duplicate_action"), + ], + ) + + result = _run_reader(log_path, "json") + trace_forest = json.loads(result.stdout) + warning_codes = {warning["code"] for warning in trace_forest["warnings"]} + + assert trace_forest["roots"] == ["root"] + assert trace_forest["nodes"]["root"]["actions"] == [ + {"name": "valid_action", "children": ["valid-child"]} + ] + assert "unlinked" in trace_forest["nodes"] + assert "missing" in trace_forest["nodes"] + assert "duplicate" not in trace_forest["nodes"] + assert warning_codes == { + "DUPLICATE_EVENT_ID", + "MISSING_PARENT", + "UNLINKED_EVENT", + } + assert "DUPLICATE_EVENT_ID" in result.stderr + assert "MISSING_PARENT" in result.stderr + assert "UNLINKED_EVENT" in result.stderr diff --git a/python/flink_agents/e2e_tests/e2e_tests_integration/python_event_logging_test.py b/python/flink_agents/e2e_tests/e2e_tests_integration/python_event_logging_test.py index 2aade7d80..52fb9831b 100644 --- a/python/flink_agents/e2e_tests/e2e_tests_integration/python_event_logging_test.py +++ b/python/flink_agents/e2e_tests/e2e_tests_integration/python_event_logging_test.py @@ -18,6 +18,8 @@ import json import os import shutil +import subprocess +import sys import sysconfig import tempfile from pathlib import Path @@ -42,6 +44,9 @@ os.environ["PYTHONPATH"] = sysconfig.get_paths()["purelib"] +_REPO_ROOT = Path(__file__).resolve().parents[4] +_TRACE_TREE_READER = _REPO_ROOT / "tools" / "reconstruct_trace_tree.py" + class InputKeySelector(KeySelector): """Key selector for input data.""" @@ -206,6 +211,81 @@ def _read_log_records(event_log_dir: Path) -> list[dict]: return records +def _run_trace_tree_reader( + event_log_dir: Path, output_format: str +) -> subprocess.CompletedProcess[str]: + """Run the public Trace Tree reader CLI against Runtime Event Logs.""" + result = subprocess.run( + [ + sys.executable, + str(_TRACE_TREE_READER), + str(event_log_dir), + "--format", + output_format, + ], + check=False, + capture_output=True, + text=True, + timeout=30, + ) + assert result.returncode == 0, result.stderr + return result + + +def test_event_lineage_reconstructs_trace_trees_from_runtime_logs( + tmp_path: Path, +) -> None: + """Test a real PyFlink job produces logs consumable by the Trace Tree reader.""" + event_log_dir = _run_event_logging_pipeline(tmp_path) + records = _read_log_records(event_log_dir) + input_event_ids = { + record["event"]["id"] + for record in records + if record["eventType"] == InputEvent.EVENT_TYPE + } + output_event_ids = { + record["event"]["id"] + for record in records + if record["eventType"] == OutputEvent.EVENT_TYPE + } + + json_result = _run_trace_tree_reader(event_log_dir, "json") + trace_forest = json.loads(json_result.stdout) + text_result = _run_trace_tree_reader(event_log_dir, "text") + + assert input_event_ids + assert len(output_event_ids) == len(input_event_ids) + assert set(trace_forest["roots"]) == input_event_ids + assert set(trace_forest["nodes"]) == input_event_ids | output_event_ids + assert trace_forest["warnings"] == [] + assert json_result.stderr == "" + + for root_id in trace_forest["roots"]: + root = trace_forest["nodes"][root_id] + assert root["eventType"] == InputEvent.EVENT_TYPE + assert root["upstreamEventId"] is None + assert root["upstreamActionName"] is None + assert len(root["actions"]) == 1 + + action_node = root["actions"][0] + assert action_node["name"] == "process_input" + assert len(action_node["children"]) == 1 + + child_id = action_node["children"][0] + child = trace_forest["nodes"][child_id] + assert child_id in output_event_ids + assert child["eventType"] == OutputEvent.EVENT_TYPE + assert child["upstreamEventId"] == root_id + assert child["upstreamActionName"] == "process_input" + assert child["actions"] == [] + + assert text_result.stdout.count("Trace Tree ") == len(input_event_ids) + assert text_result.stdout.count("[Action: process_input]") == len(input_event_ids) + for event_id in input_event_ids | output_event_ids: + assert f"({event_id})" in text_result.stdout + assert text_result.stderr == "" + + def test_event_log_verbose_level(tmp_path: Path) -> None: """Test that VERBOSE log level writes events without truncation.""" event_log_dir = _run_event_logging_pipeline( diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/EventLogRecordJsonSerdeTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/EventLogRecordJsonSerdeTest.java index 1f7a1192f..a77b079ea 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/EventLogRecordJsonSerdeTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/EventLogRecordJsonSerdeTest.java @@ -161,6 +161,9 @@ void testDeserializeOutputEvent() throws Exception { void testDeserializeCustomEvent() throws Exception { // Given CustomTestEvent originalEvent = new CustomTestEvent("custom data", 42, true); + UUID upstreamEventId = UUID.randomUUID(); + originalEvent.setUpstreamEventId(upstreamEventId); + originalEvent.setUpstreamActionName("custom_action"); EventContext originalContext = new EventContext(originalEvent); EventLogRecord originalRecord = new EventLogRecord(originalContext, originalEvent); String json = objectMapper.writeValueAsString(originalRecord); @@ -175,6 +178,9 @@ void testDeserializeCustomEvent() throws Exception { assertEquals("custom data", customEvent.getCustomData()); assertEquals(42, customEvent.getCustomNumber()); assertTrue(customEvent.isCustomFlag()); + assertEquals(originalEvent.getId(), customEvent.getId()); + assertEquals(upstreamEventId, customEvent.getUpstreamEventId()); + assertEquals("custom_action", customEvent.getUpstreamActionName()); } @Test @@ -197,6 +203,25 @@ void testRoundTripSerialization() throws Exception { assertEquals(originalEvent.getInput(), deserializedEvent.getInput()); } + @Test + void testRoundTripLineageFields() throws Exception { + UUID upstreamEventId = UUID.randomUUID(); + Event originalEvent = new Event("ChildEvent"); + originalEvent.setUpstreamEventId(upstreamEventId); + originalEvent.setUpstreamActionName("child_action"); + EventLogRecord originalRecord = + new EventLogRecord(new EventContext(originalEvent), originalEvent); + + String json = objectMapper.writeValueAsString(originalRecord); + EventLogRecord deserializedRecord = objectMapper.readValue(json, EventLogRecord.class); + + JsonNode eventNode = objectMapper.readTree(json).get("event"); + assertEquals(upstreamEventId.toString(), eventNode.get("upstreamEventId").asText()); + assertEquals("child_action", eventNode.get("upstreamActionName").asText()); + assertEquals(upstreamEventId, deserializedRecord.getEvent().getUpstreamEventId()); + assertEquals("child_action", deserializedRecord.getEvent().getUpstreamActionName()); + } + @Test void testSerializeUnifiedEvent() throws Exception { // Given - a unified event with user-defined type @@ -281,9 +306,7 @@ private CustomTestEvent(UUID id, Map attributes) { public static CustomTestEvent fromEvent(Event event) { CustomTestEvent result = new CustomTestEvent(event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/FileEventLoggerTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/FileEventLoggerTest.java index 9156aee72..92dd5a6d0 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/FileEventLoggerTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/FileEventLoggerTest.java @@ -174,6 +174,9 @@ void testAppendWithCustomEvent() throws Exception { // Given logger.open(openParams); TestCustomEvent customEvent = new TestCustomEvent("custom data", 42); + UUID upstreamEventId = UUID.randomUUID(); + customEvent.setUpstreamEventId(upstreamEventId); + customEvent.setUpstreamActionName("custom_action"); EventContext context = new EventContext(customEvent); // When @@ -203,6 +206,9 @@ void testAppendWithCustomEvent() throws Exception { TestCustomEvent.fromEvent(deserializedRecord.getEvent()); assertEquals("custom data", deserializedEvent.getCustomData()); assertEquals(42, deserializedEvent.getCustomNumber()); + assertEquals(customEvent.getId(), deserializedEvent.getId()); + assertEquals(upstreamEventId, deserializedEvent.getUpstreamEventId()); + assertEquals("custom_action", deserializedEvent.getUpstreamActionName()); } @Test @@ -566,9 +572,7 @@ private TestCustomEvent(UUID id, Map attributes) { public static TestCustomEvent fromEvent(Event event) { TestCustomEvent result = new TestCustomEvent(event.getId(), new HashMap<>(event.getAttributes())); - if (event.hasSourceTimestamp()) { - result.setSourceTimestamp(event.getSourceTimestamp()); - } + result.copyFrameworkMetadataFrom(event); return result; } diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/JsonTruncatorTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/JsonTruncatorTest.java index bd05b9a10..504b2d342 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/JsonTruncatorTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/eventlog/JsonTruncatorTest.java @@ -138,9 +138,12 @@ void testProtectedFields() { ObjectNode node = MAPPER.createObjectNode(); // Protected fields should never be truncated node.put("eventType", "org.apache.flink.agents.api.event.ChatRequestEvent"); + node.put("type", "org.apache.flink.agents.api.event.ChatRequestEvent"); node.put("id", "a-very-long-identifier-that-exceeds-the-limit"); + node.put("upstreamEventId", "a-very-long-upstream-event-identifier"); + node.put("upstreamActionName", "a-very-long-action-name"); ObjectNode attributes = MAPPER.createObjectNode(); - attributes.put("key", "value"); + attributes.put("key", "business payload should be truncated"); attributes.set("nested", MAPPER.createObjectNode().put("deep", "data")); node.set("attributes", attributes); // Non-protected field should be truncated @@ -152,10 +155,16 @@ void testProtectedFields() { // Protected fields remain untouched assertThat(node.get("eventType").asText()) .isEqualTo("org.apache.flink.agents.api.event.ChatRequestEvent"); + assertThat(node.get("type").asText()) + .isEqualTo("org.apache.flink.agents.api.event.ChatRequestEvent"); assertThat(node.get("id").asText()) .isEqualTo("a-very-long-identifier-that-exceeds-the-limit"); + assertThat(node.get("upstreamEventId").asText()) + .isEqualTo("a-very-long-upstream-event-identifier"); + assertThat(node.get("upstreamActionName").asText()).isEqualTo("a-very-long-action-name"); assertThat(node.get("attributes").isObject()).isTrue(); - assertThat(node.get("attributes").get("key").asText()).isEqualTo("value"); + assertThat(node.get("attributes").get("key").get("truncatedString").asText()) + .isEqualTo("busin..."); // Non-protected field is truncated assertThat(node.get("content").get("truncatedString")).isNotNull(); } 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 22c0072f8..26ff8eb2e 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 @@ -17,6 +17,7 @@ */ package org.apache.flink.agents.runtime.operator; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.flink.agents.api.Event; import org.apache.flink.agents.api.EventContext; @@ -34,6 +35,7 @@ import org.apache.flink.agents.plan.JavaFunction; import org.apache.flink.agents.plan.actions.Action; import org.apache.flink.agents.runtime.actionstate.ActionState; +import org.apache.flink.agents.runtime.actionstate.ActionStateSerde; import org.apache.flink.agents.runtime.actionstate.CallResult; import org.apache.flink.agents.runtime.actionstate.InMemoryActionStateStore; import org.apache.flink.agents.runtime.eventlog.FileEventLogger; @@ -50,6 +52,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.io.IOException; import java.lang.reflect.Field; import java.util.Collections; import java.util.HashMap; @@ -581,6 +584,36 @@ agentPlanWithStateStore, true, new InMemoryActionStateStore(false)), } } + @Test + void testCompletedActionStatePersistsOutputEventLineage() throws Exception { + AgentPlan agentPlan = TestAgent.getAgentPlan(false); + SerializingActionStateStore actionStateStore = new SerializingActionStateStore(); + + try (KeyedOneInputStreamOperatorTestHarness testHarness = + new KeyedOneInputStreamOperatorTestHarness<>( + new ActionExecutionOperatorFactory<>(agentPlan, true, actionStateStore), + (KeySelector) value -> value, + TypeInformation.of(Long.class))) { + testHarness.open(); + ActionExecutionOperator operator = + (ActionExecutionOperator) testHarness.getOperator(); + + testHarness.processElement(new StreamRecord<>(3L)); + operator.waitInFlightEventsFinished(); + } + + assertThat(actionStateStore.getCompletedStateBytes()).hasSize(2); + for (Map.Entry entry : + actionStateStore.getCompletedStateBytes().entrySet()) { + JsonNode state = OBJECT_MAPPER.readTree(entry.getValue()); + JsonNode outputEvent = state.path("outputEvents").get(0); + + assertThat(outputEvent.path("upstreamEventId").asText()) + .isEqualTo(state.path("taskEvent").path("id").asText()); + assertThat(outputEvent.path("upstreamActionName").asText()).isEqualTo(entry.getKey()); + } + } + @Test void testActionStateStoreStateManagement() throws Exception { AgentPlan agentPlanWithStateStore = TestAgent.getAgentPlan(false); @@ -748,6 +781,16 @@ agentPlanWithStateStore, true, new InMemoryActionStateStore(false)), testHarness.processElement(new StreamRecord<>(inputValue)); operator.waitInFlightEventsFinished(); } + + for (Map states : actionStateStore.getKeyedActionStates().values()) { + for (ActionState state : states.values()) { + for (Event outputEvent : state.getOutputEvents()) { + outputEvent.setUpstreamEventId(null); + outputEvent.setUpstreamActionName(null); + } + } + } + try (KeyedOneInputStreamOperatorTestHarness testHarness = new KeyedOneInputStreamOperatorTestHarness<>( new ActionExecutionOperatorFactory<>( @@ -772,6 +815,18 @@ agentPlanWithStateStore, true, new InMemoryActionStateStore(false)), // The action state store should only have one entry assertThat(actionStateStore.getKeyedActionStates().get(String.valueOf(inputValue))) .hasSize(2); + + for (ActionState state : + actionStateStore + .getKeyedActionStates() + .get(String.valueOf(inputValue)) + .values()) { + Event outputEvent = state.getOutputEvents().get(0); + String expectedActionName = + state.getTaskEvent() instanceof InputEvent ? "action1" : "action2"; + assertThat(outputEvent.getUpstreamEventId()).isNotNull(); + assertThat(outputEvent.getUpstreamActionName()).isEqualTo(expectedActionName); + } } } @@ -2154,6 +2209,27 @@ private List getPrunedSeqNums() { } } + private static class SerializingActionStateStore extends InMemoryActionStateStore { + private final Map completedStateBytes = new HashMap<>(); + + private SerializingActionStateStore() { + super(false); + } + + @Override + public void put(Object key, long seqNum, Action action, Event event, ActionState state) + throws IOException { + if (state.isCompleted()) { + completedStateBytes.put(action.getName(), ActionStateSerde.serialize(state)); + } + super.put(key, seqNum, action, event, state); + } + + private Map getCompletedStateBytes() { + return completedStateBytes; + } + } + private static void assertMailboxSizeAndRun(TaskMailbox mailbox, int expectedSize) throws Exception { assertThat(mailbox.size()).isEqualTo(expectedSize); diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/EventLineageReconstructionTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/EventLineageReconstructionTest.java new file mode 100644 index 000000000..72e26fa44 --- /dev/null +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/EventLineageReconstructionTest.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.operator; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.flink.agents.api.InputEvent; +import org.apache.flink.agents.api.OutputEvent; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.api.logger.EventLogLevel; +import org.apache.flink.agents.api.logger.LoggerType; +import org.apache.flink.agents.plan.AgentConfiguration; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.runtime.eventlog.EventLogRecord; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Verifies that the event log alone contains the minimal causal chain between Events. */ +class EventLineageReconstructionTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + @Test + void recordsMinimalEventLineageInTheEventLog(@TempDir Path logDir) throws Exception { + AgentPlan agentPlan = + ActionExecutionOperatorTest.TestAgent.getAgentPlanWithConfig( + fileLoggerConfig(logDir)); + + try (KeyedOneInputStreamOperatorTestHarness harness = + new KeyedOneInputStreamOperatorTestHarness<>( + new ActionExecutionOperatorFactory<>(agentPlan, true), + (KeySelector) value -> value, + TypeInformation.of(Long.class))) { + harness.open(); + harness.processElement(new StreamRecord<>(1L)); + ((ActionExecutionOperator) harness.getOperator()) + .waitInFlightEventsFinished(); + } + + Map recordsByType = readRecordsByType(logDir); + EventLogRecord inputRecord = + MAPPER.treeToValue(recordsByType.get(InputEvent.EVENT_TYPE), EventLogRecord.class); + EventLogRecord outputRecord = + MAPPER.treeToValue(recordsByType.get(OutputEvent.EVENT_TYPE), EventLogRecord.class); + JsonNode input = recordsByType.get(InputEvent.EVENT_TYPE).get("event"); + JsonNode middle = + recordsByType + .get(ActionExecutionOperatorTest.TestAgent.MiddleEvent.EVENT_TYPE) + .get("event"); + JsonNode output = recordsByType.get(OutputEvent.EVENT_TYPE).get("event"); + + assertThat(recordsByType).hasSize(3); + assertThat(inputRecord.getEvent().getType()).isEqualTo(InputEvent.EVENT_TYPE); + assertThat(outputRecord.getEvent().getType()).isEqualTo(OutputEvent.EVENT_TYPE); + assertThat(input.has("upstreamEventId")).isFalse(); + assertThat(input.has("upstreamActionName")).isFalse(); + + assertThat(middle.get("upstreamEventId").isTextual()).isTrue(); + assertThat(middle.get("upstreamActionName").isTextual()).isTrue(); + assertThat(middle.get("upstreamEventId").asText()).isEqualTo(input.get("id").asText()); + assertThat(middle.get("upstreamActionName").asText()).isEqualTo("action1"); + + assertThat(output.get("upstreamEventId").isTextual()).isTrue(); + assertThat(output.get("upstreamActionName").isTextual()).isTrue(); + assertThat(output.get("upstreamEventId").asText()).isEqualTo(middle.get("id").asText()); + assertThat(output.get("upstreamActionName").asText()).isEqualTo("action2"); + + assertThat(middle.get("attributes").has("upstreamEventId")).isFalse(); + assertThat(middle.get("attributes").has("upstreamActionName")).isFalse(); + } + + private static AgentConfiguration fileLoggerConfig(Path logDir) { + AgentConfiguration config = new AgentConfiguration(); + config.set(AgentConfigOptions.EVENT_LOGGER_TYPE, LoggerType.FILE); + config.set(AgentConfigOptions.BASE_LOG_DIR, logDir.toString()); + config.set(AgentConfigOptions.EVENT_LOG_LEVEL, EventLogLevel.STANDARD); + config.set(AgentConfigOptions.EVENT_LOG_MAX_STRING_LENGTH, 3); + return config; + } + + private static Map readRecordsByType(Path logDir) throws Exception { + List lines; + try (Stream files = Files.list(logDir)) { + List logFiles = + files.filter(path -> path.getFileName().toString().endsWith(".log")) + .collect(Collectors.toList()); + assertThat(logFiles).hasSize(1); + lines = Files.readAllLines(logFiles.get(0)); + } + + Map recordsByType = new LinkedHashMap<>(); + for (String line : lines) { + JsonNode record = MAPPER.readTree(line); + recordsByType.put(record.get("eventType").asText(), record); + } + return recordsByType; + } +}