weiqingy commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3732939310
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -81,10 +92,18 @@ public Map<String, Object> getAttributes() {
return attributes;
}
+ public Map<String, Object> getAttachments() {
+ return attachments;
+ }
+
public Object getAttr(String name) {
return attributes.get(name);
}
+ public Object getAttachment(String name) {
Review Comment:
Java gets `getAttachment` here but no `setAttachment`; Python ships both
(`event.py:146`, `:150`). AGENTS.md asks that public APIs stay aligned across
Java, Python and YAML, and today a Java user reaches through
`getAttachments().put(...)` for what Python exposes directly.
Which direction fits what you have in mind, adding the Java setter or
dropping the Python one?
##########
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:
Both runtimes reject an `OutputEvent` carrying attachments before storing
them (`EventAttachmentUtils.java:46-58`, `event_attachment_utils.py:59-62`), so
there is no Java/Python gap here to close. What is left is internal: this copy,
and the one at `event.py:254`, only ever build an object `sendEvent` refuses,
and the `output_event.json` snapshots now pin that shape as a fixture.
What is the intended contract for attachments on `OutputEvent`? That answer
decides whether the rejection moves or the copy does.
##########
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:
Including `attachments` here, and in `equals` at `:175`, makes them part of
event identity. The durable action-state key does not follow:
`ActionStateUtil.generateUUIDForEvent` hashes `event.getAttributes()` only, so
two events this line now distinguishes can still land on one `ActionState`.
A fan-out with durable execution on is where that shows: `ctx.sendEvent(new
Event("WorkItem", new HashMap<>(), Map.of("payload", item)))` in a loop gives
every sibling the same empty `attributes`, the same seqNum and the same action,
so one state key covers all of them. Item 1 completes, item 2's lookup returns
item 1's completed state, and `ActionExecutionOperator.java:341` skips
execution and replays item 1's output in its place.
Adding `attachments` to the key may just trade one problem for another,
since a ref's path embeds the random event id the key deliberately avoids
(`buildAttachmentPath`). I'm confident on the mechanism, less so on the odds,
since it needs `ACTION_STATE_STORE_BACKEND` set plus siblings with equal
`attributes`. Does that combination look reachable in practice?
##########
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:
I went looking for what this assertion would catch. It checks the final
`OutputEvent` payload, which is identical whether the attachment travelled
offloaded or inline, so a regression that skipped the offload entirely would
still pass. Nothing here observes a `MemoryRef` in flight.
None of the four new wiring points is covered by a test that carries an
attachment either: `RunnerContextImpl.java:154`, `JavaActionTask.java:60`,
`PythonActionExecutor.java:139-141`, `flink_runner_context.py:301`. Two even
look removable without failing anything. Without `flink_runner_context.py:301`,
Java's `sendEvent` offloads the dict instead and the output is unchanged.
Without `JavaActionTask.java:60`, nothing fails either, since no test runs a
Java action against an event carrying an attachment.
What would you want a test to pin down here? A `MemoryRef` at the send
boundary and the resolved value at the receive boundary is the shape I'd reach
for, and `ActionExecutionOperatorTest` looks like it could host it.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/memory/EventAttachmentUtils.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.memory;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.OutputEvent;
+import org.apache.flink.agents.api.context.MemoryObject;
+import org.apache.flink.agents.api.context.MemoryRef;
+import org.apache.flink.agents.api.context.RunnerContext;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+/** Stores event attachments in sensory memory while events cross action
boundaries. */
+public final class EventAttachmentUtils {
+
+ private static final String ATTACHMENT_ROOT = "__event_attachments__";
+
+ private EventAttachmentUtils() {}
+
+ /** Stores concrete attachment values and replaces them with
sensory-memory references. */
+ public static void storeEventAttachments(Event event, RunnerContext
context) throws Exception {
+ if (event.getAttachments().isEmpty()) {
+ return;
+ }
+
+ if (OutputEvent.EVENT_TYPE.equals(event.getType())) {
+ String keys =
+ event.getAttachments().keySet().stream()
+ .sorted()
+ .collect(Collectors.joining(", "));
+ throw new IllegalArgumentException(
+ "Output events cannot carry attachments: event_id="
+ + event.getId()
+ + ", event_type="
+ + event.getType()
+ + ", key="
+ + keys);
+ }
+
+ for (Map.Entry<String, Object> entry :
event.getAttachments().entrySet()) {
+ String key = entry.getKey();
+ Object value = entry.getValue();
+ if (value instanceof MemoryRef) {
+ continue;
+ }
+
+ MemoryRef reference =
+
context.getSensoryMemory().set(buildAttachmentPath(event.getId(), key), value);
+
+ event.getAttachments().put(key, reference);
+ }
+ }
+
+ /** Loads sensory-memory references in place before a Java action is
invoked. */
+ public static void loadEventAttachments(Event event, RunnerContext
context) throws Exception {
+ for (Map.Entry<String, Object> entry :
event.getAttachments().entrySet()) {
+ Object value = entry.getValue();
+ if (!(value instanceof MemoryRef)) {
+ continue;
+ }
+ MemoryRef reference = (MemoryRef) value;
+
+ MemoryObject attachment =
context.getSensoryMemory().get(reference);
+ if (attachment == null) {
+ throw new IllegalStateException(
+ "Event attachment does not exist in sensory memory: "
+ + reference.getPath());
+ }
+ event.getAttachments().put(entry.getKey(), attachment.getValue());
Review Comment:
This writes the resolved value back into `event.getAttachments()`, and that
`event` is `ActionTask.event`, the instance the runtime owns
(`JavaActionTask.java:60`). Python resolves against a per-invocation copy
instead (`PythonActionExecutor.java:138-141`), so its event keeps the refs.
With durable execution on, that undoes the offload: `maybeInitActionState`
stores the live event before `invoke()` runs
(`DurableExecutionManager.java:219`) and `ActionState` holds it by reference,
so every later persist writes the payload inline instead of the ref. There may
be a second effect on a heap backend, where
`ActionExecutionOperator.java:271-273` shares one `event` across sibling tasks,
though that depends on ListState value semantics I did not run.
Was the in-place write deliberate, or would resolving into a copy the action
owns work here?
--
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]
