weiqingy commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3890876674
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -135,11 +171,12 @@ public boolean equals(Object o) {
Event other = (Event) o;
return Objects.equals(this.id, other.id)
&& Objects.equals(this.getType(), other.getType())
- && Objects.equals(this.attributes, other.attributes);
+ && Objects.equals(this.attributes, other.attributes)
+ && Objects.equals(this.attachments, other.attachments);
}
@Override
public int hashCode() {
- return Objects.hash(id, getType(), attributes);
+ return Objects.hash(id, getType(), attributes, attachments);
Review Comment:
Yes, an occurrence id is the right direction. But it cannot be
`Event.getId()`, and it cannot be a fresh random one either.
`EventRouter.wrapToInputEvent` builds the root event with `new
InputEvent(input)`, which ends in `UUID.randomUUID()`. A record replayed from
the source after a checkpoint gets a new id. So an id-based key would never
find the old state, and every replayed record would re-run its whole chain.
That is the case durable execution exists to avoid.
`ActionStateUtilTest.testGenerateKeyConsistency` has pinned this since #138:
two `InputEvent("same-input")` with different ids must produce the same key.
So the id has to do two things at once. It has to be unique per occurrence
inside one `(businessKey, seqNum, action)`, and it has to come out the same
when the run is replayed from the source. A random id gives you the first and
loses the second. Attribute content gives you the second and loses the first.
Something derived from the event's position in the run gives you both.
There may already be a pattern to borrow.
`RunnerContextImpl.matchNextOrClearSubsequentCallResult` identifies durable
calls by their ordinal index in a persisted list, and uses `functionId +
argsDigest` only to detect that replay diverged, clearing the later results
when it does. Would that shape fit events too, with an ordinal in the key and
the content hash demoted to a validation check?
If you go that way, an ordinal is the cheap option, but it is only
replay-stable if dispatch order is deterministic across a restart, and I have
not checked whether that holds once continuations and async calls interleave. A
lineage path (parent occurrence, action name, index in that action's output
list) sidesteps that question. `upstreamEventId` already carries the parent
link, though it would need to point at the parent's derived identity rather
than its random id.
One thing that may take some pressure off this PR: the collision does not
need attachments. Two events with the same attributes sent to the same action
already land on one `ActionState` on main today, since `processEvent` does not
dedup and the key hashes attributes only.
Your PR makes it much easier to hit, because a fan-out that carries its
payload in an attachment leaves the attributes identical by design. But the
hole is already there. So this looks like a durable execution issue rather than
an attachments one, and probably wants its own issue and its own tests rather
than a fix squeezed in here.
Also my repro above does not compile, sorry. There is no `Event(String, Map,
Map)` constructor. The simple version is `Event e = new Event("WorkItem");
e.setAttachment("payload", item);`
##########
python/flink_agents/e2e_tests/e2e_tests_integration/event_attachments_test.py:
##########
@@ -0,0 +1,100 @@
+################################################################################
+# 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 os
+import sys
+import sysconfig
+from pathlib import Path
+from typing import Any
+
+from pyflink.common import Configuration
+from pyflink.datastream import KeySelector, StreamExecutionEnvironment
+
+from flink_agents.api.agents.agent import Agent
+from flink_agents.api.decorators import action
+from flink_agents.api.events.event import Event, InputEvent, OutputEvent
+from flink_agents.api.execution_environment import AgentsExecutionEnvironment
+from flink_agents.api.runner_context import RunnerContext
+
+current_dir = Path(__file__).parent
+os.environ["PYTHONPATH"] = (
+ f"{current_dir.parent.parent.parent}:{sysconfig.get_paths()['purelib']}"
+)
+
+
+class _KeySelector(KeySelector):
+ def get_key(self, value: dict[str, Any]) -> str:
+ return str(value["key"])
+
+
+class EventAttachmentsAgent(Agent):
+ @action(InputEvent.EVENT_TYPE)
+ @staticmethod
+ def send_attachment(event: Event, ctx: RunnerContext) -> None:
+ value = InputEvent.from_event(event).input
+ ctx.send_event(
+ Event(
+ type="AttachmentStep",
+ attributes={"kind": "inline"},
+ attachments={
+ "payload": {
+ "value": value,
+ "items": [1, 2, 3],
+ }
+ },
+ )
+ )
+
+ @action("AttachmentStep")
+ @staticmethod
+ def receive_attachment(event: Event, ctx: RunnerContext) -> None:
+ print(f"received attachments: {event.attachments}")
+ ctx.send_event(
+ OutputEvent(
+ output={
+ "kind": event.get_attr("kind"),
+ "payload": event.get_attachment("payload"),
+ }
+ )
+ )
+
+
+def test_python_event_attachments_roundtrip_on_flink() -> None:
+ config = Configuration()
+ config.set_string("python.pythonpath", os.environ["PYTHONPATH"])
+ env = StreamExecutionEnvironment.get_execution_environment(config)
+ env.set_python_executable(sys.executable)
+ env.set_parallelism(1)
+ input_stream = env.from_collection(
+ [{"key": "k1", "value": {"message": "hello"}}]
+ )
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
+ output = (
+ agents_env.from_datastream(input_stream, _KeySelector())
+ .apply(EventAttachmentsAgent())
+ .to_datastream()
+ )
+
+ assert list(output.execute_and_collect()) == [
Review Comment:
Thanks, this covers it. I deleted the offload call and then moved it after
serialization, and the Python test failed both times, so it pins the boundary
and not just the payload.
`PythonActionExecutor.java:139-140` is the one wiring point still reached
only by the e2e test, which `ut.sh` skips by default. Is e2e the right level
for the bridge, or would a small unit test there be worth having?
##########
api/src/main/java/org/apache/flink/agents/api/OutputEvent.java:
##########
@@ -54,6 +54,7 @@ public OutputEvent(
*/
public static OutputEvent fromEvent(Event event) {
OutputEvent result = new OutputEvent(event.getId(), new
HashMap<>(event.getAttributes()));
+ result.getAttachments().putAll(event.getAttachments());
Review Comment:
Resolved, so I am closing this one out. `fromEvent` now rejects attachments,
Python raises the same message, both `output_event.json` snapshots are clean,
and there are tests on both sides.
--
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]
