joeyutong commented on code in PR #923:
URL: https://github.com/apache/flink-agents/pull/923#discussion_r3643415986
##########
python/flink_agents/api/events/event.py:
##########
@@ -96,35 +115,44 @@ 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)
- 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_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_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
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:
Review Comment:
The Java and Python helpers currently have different contracts: Python
returns a copy and includes the Event ID, while Java mutates the current object
and does not copy the ID. Also, “framework metadata” describes ownership rather
than the Event-domain meaning. Could the names and contracts be aligned around
preserving/reconstructing the same Event occurrence?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -454,6 +456,13 @@ private void processActionTaskForKey(Object key) throws
Exception {
}
}
+ private static void setOutputEventLineage(ActionTask actionTask,
List<Event> outputEvents) {
Review Comment:
This rule describes how a concrete Action invocation finalizes its output
Events, rather than Operator scheduling. Could it be owned by `ActionTask`,
which already holds the triggering Event and Action? That would keep the
Operator independent of lineage field details and provide one place to reject
`outputEvent.id == triggerEvent.id`; currently `ctx.sendEvent(event)` can
create a self-loop and mutate the triggering Event shared by sibling
ActionTasks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
