You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "JinkunLiu (via GitHub)" <gi...@apache.org> on 2026/08/02 09:42:26 UTC
[PR] Feature/476 auto resolve memory ref [flink-agents]
JinkunLiu opened a new pull request, #950:
URL: https://github.com/apache/flink-agents/pull/950
<!--
* Thank you very much for contributing to Flink Agents.
* Please add the relevant components in the PR title. E.g., [api], [runtime], [java], [python], [hotfix], etc.
-->
<!-- Please link the PR to the relevant issue(s). Hotfix doesn't need this. -->
Linked issue: #476
### Purpose of change
Add event attachments with automatic MemoryRef wrapping and resolution
<!-- What is the purpose of this change? -->
### Tests
<!-- How is this change verified? -->
- Java and Python unit tests for resolving and restoring event attachments.
- Event serialization and deserialization tests.
- Cross-language snapshot tests covering Java-to-Python and Python-to-Java event conversion.
- End-to-end tests covering actions that consume and produce MemoryRef attachments.
### API
<!-- Does this change touches any public APIs? -->
- Add public APIs for event attachments in both Java and Python
- Add JSON serialization/deserialization support to MemoryRef.
### Documentation
<!-- Do not remove this section. Check the proper box only. -->
- [x] `doc-needed` <!-- Your PR changes impact docs -->
Add docs after review～
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-included` <!-- Your PR already contains the necessary documentation updates -->
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3837774670
##########
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:
Thanks for point this problem, the unconditional copy is a performance issue.
In commit [28da154e](https://github.com/apache/flink-agents/pull/950/changes/28da154ecf976dd69399290fc9db86987d8b49a8):
- skip the copy when the event contains no MemoryRef attachments;
- Use job serializer config for event copies;
- add a new assert at `resolvesAttachmentsWithoutMutatingDurableEvent` test
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #950:
URL: https://github.com/apache/flink-agents/pull/950#issuecomment-5165724783
Thanks for taking this on @JinkunLiu. I'll review it ASAP.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3791296469
##########
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);
Review Comment:
Thanks for you point that,Fix it in [e1242310](https://github.com/apache/flink-agents/pull/950/changes/e124231097987aca583715211c4504d0757c4d60)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3912477652
##########
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:
Thanks for the detailed analysis. I created #1084 to track this separately. Since the collision already exists without attachments and likely requires a replay-stable occurrence identity, I agree that it should not be addressed in this PR.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 merged PR #950:
URL: https://github.com/apache/flink-agents/pull/950
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3943920835
##########
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:
Yes, I think it’s worthwhile, and I have added a small unit test in [3584c7](https://github.com/apache/flink-agents/pull/950/changes/3584c76f4971781ab458fb00c8278bf56f08a169)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3725525928
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -125,7 +149,19 @@ public static Event fromEvent(Event event) {
* @throws IOException if JSON parsing fails or the 'type' field is missing or empty
*/
public static Event fromJson(String json) throws IOException {
- return MAPPER.readValue(json, Event.class);
+ Event event = MAPPER.readValue(json, Event.class);
+ for (Map.Entry<String, Object> entry : event.getAttachments().entrySet()) {
Review Comment:
**[P1] Preserve `MemoryRef` when restoring `ActionState`**
Thanks for adding JSON support for `MemoryRef`. This conversion only runs when `Event.fromJson()` is called explicitly. Durable recovery instead deserializes the enclosing `ActionState` directly through `ActionStateSerde`, so attachment values declared as `Object` are restored as `LinkedHashMap` rather than `MemoryRef`. `loadEventAttachments()` then skips them, and the recovered action receives the reference-shaped map instead of the original payload.
A possible implementation sketch would be to bind a content deserializer directly to the `attachments` values:
```java
@JsonCreator
public Event(
// ...
@JsonProperty("attachments")
@JsonDeserialize(contentUsing = AttachmentValueDeserializer.class)
Map<String, Object> attachments) {
// ...
}
static final class AttachmentValueDeserializer extends JsonDeserializer<Object> {
@Override
public Object deserialize(JsonParser p, DeserializationContext ctxt)
throws IOException {
JsonNode node = p.getCodec().readTree(p);
if (node.isObject()
&& "memory_ref".equals(node.path("@type").asText())) {
return p.getCodec().treeToValue(node, MemoryRef.class);
}
return p.getCodec().treeToValue(node, Object.class);
}
}
```
The `MemoryRef` serializer would emit a language-neutral discriminator, for example `{"@type": "memory_ref", "memory_type": "sensory", "path": "..."}`. Merely adding the discriminator to the JSON is not sufficient: because the declared map value type is still `Object`, Jackson also needs this property-level deserialization hook (or an equivalent central hook) to select `MemoryRef.Deserializer`. This should use the same discriminator on the Python side, and an `ActionStateSerde` round-trip test should verify that the restored attachment is still a `MemoryRef`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3743281042
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -125,7 +149,19 @@ public static Event fromEvent(Event event) {
* @throws IOException if JSON parsing fails or the 'type' field is missing or empty
*/
public static Event fromJson(String json) throws IOException {
- return MAPPER.readValue(json, Event.class);
+ Event event = MAPPER.readValue(json, Event.class);
+ for (Map.Entry<String, Object> entry : event.getAttachments().entrySet()) {
Review Comment:
Done，with this [commit](https://github.com/apache/flink-agents/pull/950/changes/9bde2748ceb17117521081f91c21709a9ddb9c90)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3890875875
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java:
##########
@@ -144,6 +145,52 @@ void testExecuteAgent() throws Exception {
}
}
+ @Test
+ void testJavaEventAttachmentsAreOffloadedAndResolvedBetweenActions() throws Exception {
+ InMemoryActionStateStore actionStateStore = new InMemoryActionStateStore(false);
+ AgentPlan agentPlan = TestAgent.getEventAttachmentAgentPlan();
+ try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new ActionExecutionOperatorFactory<>(agentPlan, true, actionStateStore),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class))) {
+ testHarness.open();
+ ActionExecutionOperator<Long, Object> operator =
+ (ActionExecutionOperator<Long, Object>) testHarness.getOperator();
+
+ List<Event> eventsAtSendBoundary = new ArrayList<>();
+ operator.getEventRouter()
+ .addEventListener(
+ (context, event) -> {
+ if (TestAgent.ATTACHMENT_EVENT_TYPE.equals(event.getType())) {
+ eventsAtSendBoundary.add(event);
+ }
+ });
+
+ testHarness.processElement(new StreamRecord<>(1L));
+ operator.waitInFlightEventsFinished();
+
+ assertThat(eventsAtSendBoundary).hasSize(1);
+ Event runtimeEvent = eventsAtSendBoundary.get(0);
+ Object reference = runtimeEvent.getAttachment(TestAgent.ATTACHMENT_KEY);
+ assertThat(reference).isInstanceOf(MemoryRef.class);
+ List<StreamRecord<Object>> recordOutput =
+ (List<StreamRecord<Object>>) testHarness.getRecordOutput();
+ assertThat(recordOutput).hasSize(1);
+ assertThat(recordOutput.get(0).getValue()).isEqualTo(Map.of("value", 1L));
+
+ ActionState actionState =
+ actionStateStore.get(
+ 1L,
+ 0L,
+ agentPlan.getActions().get("receiveEventAttachment"),
+ runtimeEvent);
+ assertThat(actionState).isNotNull();
+ assertThat(actionState.getTaskEvent().getAttachment(TestAgent.ATTACHMENT_KEY))
+ .isSameAs(reference);
Review Comment:
I think this one can never fail. `actionState.getTaskEvent()` is the same object as `runtimeEvent`.
`processEvent` passes one `event` instance both to `eventRouter.notifyEventProcessed` (`ActionExecutionOperator.java:298`) and to `createActionTask` (`:328`). `ActionTask.event` is `final` (`ActionTask.java:54`), and the resolved copy goes to a different field, `JavaActionTask.invocationEvent` (`JavaActionTask.java:45`, assigned at `:75`), so the task event is never replaced. `new ActionState(event)` keeps that same reference (`DurableExecutionManager.java:231`), and `InMemoryActionStateStore` is a plain `HashMap` with no serialization. So this compares `reference` with itself.
Line 176 is the check that does the work. It already fails if resolution leaks back into the runtime event.
There is also a shape question. The identity only holds because this store does not serialize. A Kafka or Fluss store would round-trip the event and break `isSameAs` even though the contract still holds. `ActionStateSerdeTest:96-97` uses `assertInstanceOf` plus `assertEquals` for the same idea. Would that fit better here?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/memory/EventAttachmentUtils.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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 org.apache.flink.api.common.typeutils.TypeSerializer;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Map;
+import java.util.Objects;
+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) {
+ MemoryRef reference = (MemoryRef) value;
+ if (!MemoryObject.MemoryType.SENSORY.equals(reference.getType())) {
+ throw new IllegalArgumentException(
+ "Event attachments must use sensory memory references: event_id="
+ + event.getId()
+ + ", event_type="
+ + event.getType()
+ + ", key="
+ + key
+ + ", memory_type="
+ + reference.getType());
+ }
+ continue;
+ }
+
+ MemoryRef reference =
+ context.getSensoryMemory().set(buildAttachmentPath(event.getId(), key), value);
+
+ event.getAttachments().put(key, reference);
+ }
+ }
+
+ /** Returns an Event with sensory-memory references resolved on a copy when necessary. */
+ public static Event loadEventAttachments(
+ Event event, RunnerContext context, TypeSerializer<Event> eventSerializer)
+ throws Exception {
+ boolean requiresResolution =
+ event.getAttachments().values().stream().anyMatch(MemoryRef.class::isInstance);
+ if (!requiresResolution) {
+ return event;
+ }
+
+ Event actionEvent =
+ Objects.requireNonNull(
+ eventSerializer,
+ "Event serializer is required to resolve attachment references.")
+ .copy(event);
+ for (Map.Entry<String, Object> entry : actionEvent.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(
Review Comment:
nit: Python covers this branch with `test_load_rejects_missing_event_attachment` (`test_event_attachment_utils.py:133`), but I could not find a Java twin. `EventAttachmentUtilsTest` has seven tests and none reach the null path. Worth a short one next to `loadsEventAttachments`?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3836719754
##########
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:
Thanks for point that. I wasn’t familiar with durable execution, so I missed that resolving the attachments also modified the event stored in ActionState
In [ba79e7dd](https://github.com/apache/flink-agents/pull/950/changes/ba79e7dd57c538ff1e01c258ed1595ccfbdc54df), I changed Java attachment resolution to operate on an copy of the event instead of mutating the original instance. I also added a durable-execution test to verify that.
Could you please help check this is corret now? Thanks!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3912090331
##########
python/flink_agents/runtime/memory/event_attachment_utils.py:
##########
@@ -0,0 +1,104 @@
+################################################################################
+# 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.
+#################################################################################
+from __future__ import annotations
+
+import hashlib
+from typing import TYPE_CHECKING, Any
+
+from flink_agents.api.events.event import OutputEvent
+from flink_agents.api.memory_object import MemoryType, validate_memory_value
+from flink_agents.api.memory_reference import MemoryRef
+
+if TYPE_CHECKING:
+ from uuid import UUID
+
+ from flink_agents.api.events.event import Event
+ from flink_agents.api.runner_context import RunnerContext
+
+_ATTACHMENT_ROOT = "__event_attachments__"
+
+
+class EventAttachmentError(RuntimeError):
+ """Raised when an Event attachment cannot be stored or loaded."""
+
+
+def _hash_attachment_key(key: str) -> str:
+ return hashlib.sha256(key.encode("UTF-8")).hexdigest()
+
+
+def build_attachment_path(event_id: UUID, key: str) -> str:
+ """Build the canonical SensoryMemory path for one attachment."""
+ return f"{_ATTACHMENT_ROOT}.{event_id}.{_hash_attachment_key(key)}"
+
+
+def _attachment_context(event: Event, key: str, path: str | None = None) -> str:
+ suffix = f", path={path}" if path is not None else ""
+ return f"event_id={event.id}, event_type={event.type}, key={key}{suffix}"
+
+
+def store_event_attachments(event: Event, ctx: RunnerContext) -> None:
+ """Store concrete attachment values in SensoryMemory and replace them with refs."""
+ if not event.attachments:
+ return
+
+ if event.type == OutputEvent.EVENT_TYPE:
+ keys = ", ".join(sorted(event.attachments))
+ msg = f"Output events cannot carry attachments: {_attachment_context(event, keys)}"
+ raise EventAttachmentError(msg)
+
+ pending: list[tuple[str, str, Any]] = []
+ for key, value in event.attachments.items():
+ if isinstance(value, MemoryRef):
+ if value.memory_type != MemoryType.SENSORY:
+ msg = (
+ "Event attachments must use sensory memory references: "
+ f"{_attachment_context(event, key, value.path)}"
+ )
+ raise EventAttachmentError(msg)
+ continue
+ path = build_attachment_path(event.id, key)
+ try:
+ validate_memory_value(path, value)
+ except Exception as exc:
+ msg = f"Invalid event attachment value: {_attachment_context(event, key, path)}"
+ raise EventAttachmentError(msg) from exc
+ pending.append((key, path, value))
+
+ for key, path, value in pending:
+ event.attachments[key] = ctx.sensory_memory.set(path, value)
+
+
+def load_event_attachments(event: Event, ctx: RunnerContext) -> None:
+ """Load sensory refs in place immediately before a Python Action runs."""
+ for key, value in list(event.attachments.items()):
+ if not isinstance(value, MemoryRef):
+ continue
+
+ try:
+ attachment = ctx.sensory_memory.get(value)
+ except Exception as exc:
+ msg = f"Failed to load event attachment: {_attachment_context(event, key, value.path)}"
+ raise EventAttachmentError(msg) from exc
+
+ if attachment is None:
Review Comment:
`None` is a valid memory value, but `get()` returns `None` both for a stored `None` and for a missing path. As a result, a valid `None` attachment is incorrectly reported as missing here. Could we use `is_exist(value.path)` to distinguish these cases and add a regression test?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3943998355
##########
python/flink_agents/runtime/memory/event_attachment_utils.py:
##########
@@ -0,0 +1,104 @@
+################################################################################
+# 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.
+#################################################################################
+from __future__ import annotations
+
+import hashlib
+from typing import TYPE_CHECKING, Any
+
+from flink_agents.api.events.event import OutputEvent
+from flink_agents.api.memory_object import MemoryType, validate_memory_value
+from flink_agents.api.memory_reference import MemoryRef
+
+if TYPE_CHECKING:
+ from uuid import UUID
+
+ from flink_agents.api.events.event import Event
+ from flink_agents.api.runner_context import RunnerContext
+
+_ATTACHMENT_ROOT = "__event_attachments__"
+
+
+class EventAttachmentError(RuntimeError):
+ """Raised when an Event attachment cannot be stored or loaded."""
+
+
+def _hash_attachment_key(key: str) -> str:
+ return hashlib.sha256(key.encode("UTF-8")).hexdigest()
+
+
+def build_attachment_path(event_id: UUID, key: str) -> str:
+ """Build the canonical SensoryMemory path for one attachment."""
+ return f"{_ATTACHMENT_ROOT}.{event_id}.{_hash_attachment_key(key)}"
+
+
+def _attachment_context(event: Event, key: str, path: str | None = None) -> str:
+ suffix = f", path={path}" if path is not None else ""
+ return f"event_id={event.id}, event_type={event.type}, key={key}{suffix}"
+
+
+def store_event_attachments(event: Event, ctx: RunnerContext) -> None:
+ """Store concrete attachment values in SensoryMemory and replace them with refs."""
+ if not event.attachments:
+ return
+
+ if event.type == OutputEvent.EVENT_TYPE:
+ keys = ", ".join(sorted(event.attachments))
+ msg = f"Output events cannot carry attachments: {_attachment_context(event, keys)}"
+ raise EventAttachmentError(msg)
+
+ pending: list[tuple[str, str, Any]] = []
+ for key, value in event.attachments.items():
+ if isinstance(value, MemoryRef):
+ if value.memory_type != MemoryType.SENSORY:
+ msg = (
+ "Event attachments must use sensory memory references: "
+ f"{_attachment_context(event, key, value.path)}"
+ )
+ raise EventAttachmentError(msg)
+ continue
+ path = build_attachment_path(event.id, key)
+ try:
+ validate_memory_value(path, value)
+ except Exception as exc:
+ msg = f"Invalid event attachment value: {_attachment_context(event, key, path)}"
+ raise EventAttachmentError(msg) from exc
+ pending.append((key, path, value))
+
+ for key, path, value in pending:
+ event.attachments[key] = ctx.sensory_memory.set(path, value)
+
+
+def load_event_attachments(event: Event, ctx: RunnerContext) -> None:
+ """Load sensory refs in place immediately before a Python Action runs."""
+ for key, value in list(event.attachments.items()):
+ if not isinstance(value, MemoryRef):
+ continue
+
+ try:
+ attachment = ctx.sensory_memory.get(value)
+ except Exception as exc:
+ msg = f"Failed to load event attachment: {_attachment_context(event, key, value.path)}"
+ raise EventAttachmentError(msg) from exc
+
+ if attachment is None:
Review Comment:
Good catch，fix it in [158da4](https://github.com/apache/flink-agents/pull/950/changes/158da4a4332a428e43a39da7dc40cd450983ab8b)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3743372184
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -46,25 +48,34 @@ public class Event {
/** Unified event with user-defined type and attributes. */
public Event(String type, Map<String, Object> attributes) {
- this(UUID.randomUUID(), type, attributes);
+ this(UUID.randomUUID(), type, attributes, new HashMap<>());
}
/** Unified event with user-defined type and empty attributes. */
public Event(String type) {
this(type, new HashMap<>());
}
- @JsonCreator
public Event(
@JsonProperty("id") UUID id,
@JsonProperty("type") String type,
@JsonProperty("attributes") Map<String, Object> attributes) {
+ this(id, type, attributes, new HashMap<>());
+ }
+
+ @JsonCreator
+ public Event(
+ @JsonProperty("id") UUID id,
+ @JsonProperty("type") String type,
+ @JsonProperty("attributes") Map<String, Object> attributes,
+ @JsonProperty("attachments") Map<String, Object> attachments) {
if (type == null || type.isEmpty()) {
throw new IllegalArgumentException("Event 'type' must not be null or empty.");
}
this.id = id;
this.type = type;
this.attributes = attributes != null ? attributes : new HashMap<>();
+ this.attachments = attachments != null ? attachments : new HashMap<>();
Review Comment:
Fix it in both attachments and attributes. [e2aa09e0](https://github.com/apache/flink-agents/pull/950/changes/e2aa09e04c7b199e326c97b7d844f2b6b8a0d7e3)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3725525928
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -125,7 +149,19 @@ public static Event fromEvent(Event event) {
* @throws IOException if JSON parsing fails or the 'type' field is missing or empty
*/
public static Event fromJson(String json) throws IOException {
- return MAPPER.readValue(json, Event.class);
+ Event event = MAPPER.readValue(json, Event.class);
+ for (Map.Entry<String, Object> entry : event.getAttachments().entrySet()) {
Review Comment:
**[P1] Preserve `MemoryRef` when restoring `ActionState`**
Thanks for adding JSON support for `MemoryRef`. This conversion only runs when `Event.fromJson()` is called explicitly. Durable recovery instead deserializes the enclosing `ActionState` directly through `ActionStateSerde`, so attachment values declared as `Object` are restored as `LinkedHashMap` rather than `MemoryRef`. `loadEventAttachments()` then skips them, and the recovered action receives the reference-shaped map instead of the original payload. Could we move this conversion into the `attachments` Jackson deserialization path (for example, using an explicit discriminator plus a content deserializer) and add an `ActionStateSerde` round-trip test?
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -46,25 +48,34 @@ public class Event {
/** Unified event with user-defined type and attributes. */
public Event(String type, Map<String, Object> attributes) {
- this(UUID.randomUUID(), type, attributes);
+ this(UUID.randomUUID(), type, attributes, new HashMap<>());
}
/** Unified event with user-defined type and empty attributes. */
public Event(String type) {
this(type, new HashMap<>());
}
- @JsonCreator
public Event(
@JsonProperty("id") UUID id,
@JsonProperty("type") String type,
@JsonProperty("attributes") Map<String, Object> attributes) {
+ this(id, type, attributes, new HashMap<>());
+ }
+
+ @JsonCreator
+ public Event(
+ @JsonProperty("id") UUID id,
+ @JsonProperty("type") String type,
+ @JsonProperty("attributes") Map<String, Object> attributes,
+ @JsonProperty("attachments") Map<String, Object> attachments) {
if (type == null || type.isEmpty()) {
throw new IllegalArgumentException("Event 'type' must not be null or empty.");
}
this.id = id;
this.type = type;
this.attributes = attributes != null ? attributes : new HashMap<>();
+ this.attachments = attachments != null ? attachments : new HashMap<>();
Review Comment:
**[P2] Make the Event own a mutable attachment map**
Thanks for the update. It looks like the immutable-map issue can still occur in the current version: the constructor stores the caller's map directly, while `storeEventAttachments()` and `loadEventAttachments()` later mutate it with `put()`. A common call such as `new Event(..., Map.of("payload", value))` therefore writes the value to sensory memory and then fails with `UnsupportedOperationException`; a mutable shared map is modified behind the caller's back instead. Could we defensively copy it with `new HashMap<>(attachments)` and add a `Map.of(...)` regression test?
##########
python/flink_agents/api/events/event.py:
##########
@@ -72,11 +74,14 @@ 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.
+ attachments : Dict[str, Any]
+ Key-value data passed between actions through sensory memory.
"""
id: UUID = Field(default=None)
type: str
attributes: Dict[str, Any] = Field(default_factory=dict)
+ attachments: Dict[str, Any] = Field(default_factory=dict)
Review Comment:
**[P2] Avoid JSON-serializing raw attachments before offload**
Thanks for adding the Python attachment API. Because `attachments` is a regular Pydantic field, Event construction immediately includes the raw values in `_generate_content_based_id()`, and `validate_and_set_id()` serializes the entire Event again before `store_event_attachments()` can offload anything. This preserves the full JSON SerDe cost and also rejects valid memory payloads that are not JSON-serializable—for example, `attachments={"payload": b"\xff\x00"}` fails while constructing the Event. Could we keep raw attachments out of this generic JSON validation path and use an ID/offload strategy that does not require serializing the payload first? A regression test with non-UTF-8 bytes would help cover this.
##########
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);
Review Comment:
**[P2] Resolve existing references according to their memory type**
Thanks for centralizing attachment resolution here. `storeEventAttachments()` accepts and skips every existing `MemoryRef`, but this load path always queries sensory memory. A `SHORT_TERM` reference is therefore accepted on send and then looked up in the wrong store; the Python equivalent also replaces a missing lookup with `None` without reporting it. Could we resolve through `MemoryRef.resolve(context)` (or dispatch on `memory_type`), or alternatively reject non-sensory references explicitly when storing? The Python path should also treat a missing resolved value as an error.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3743271920
##########
python/flink_agents/api/events/event.py:
##########
@@ -72,11 +74,14 @@ 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.
+ attachments : Dict[str, Any]
+ Key-value data passed between actions through sensory memory.
"""
id: UUID = Field(default=None)
type: str
attributes: Dict[str, Any] = Field(default_factory=dict)
+ attachments: Dict[str, Any] = Field(default_factory=dict)
Review Comment:
Yes, this is a clear bug, and I have fixed it on this [commit](https://github.com/apache/flink-agents/pull/950/changes/e4ffa8eaab35730cff4bdcd15b17ad40fbca9efe).
While working on the fix, I also noticed that Python generated event IDs from an MD5 hash of serialized event content. As a result, two events with identical attributes could receive the same ID, which could cause attachment values to overwrite each other because their MemoryRef paths are derived from the event ID.
I have also updated Python to generate event IDs with uuid.uuid4(), matching Java’s UUID.randomUUID() behavior. If Java and Python intentionally used different ID-generation strategies for historical reasons, please let me know.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3743361878
##########
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:
I forget the `setAttachment ` function in java. It have been fix now
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3887720590
##########
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:
Thanks for point that. I think the case you described can occur when durable execution is enabled.
However, I do not see a simple fix that fits within this PR. Maybe we need to redesign `generateUUIDForEvent`, perhaps by introducing a new ID to identify each event occurrence is better than deriving the `Event` identity from values carried.
Do you have any suggestions on this? I would appreciate your thoughts.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #950:
URL: https://github.com/apache/flink-agents/pull/950#issuecomment-5661867376
Thanks for the update. The code changes look good to me now. The remaining items are:
- Merge or rebase the latest `main`, resolve the conflicts, update the affected tests, and rerun CI.
- Correct the `MemoryRef` API documentation, which currently refers only to Short-Term Memory, and add user-facing documentation for event attachments. The documentation work can also be handled in a separate PR.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3698599248
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -46,25 +48,34 @@ public class Event {
/** Unified event with user-defined type and attributes. */
public Event(String type, Map<String, Object> attributes) {
- this(UUID.randomUUID(), type, attributes);
+ this(UUID.randomUUID(), type, attributes, new HashMap<>());
}
/** Unified event with user-defined type and empty attributes. */
public Event(String type) {
this(type, new HashMap<>());
}
- @JsonCreator
public Event(
@JsonProperty("id") UUID id,
@JsonProperty("type") String type,
@JsonProperty("attributes") Map<String, Object> attributes) {
+ this(id, type, attributes, new HashMap<>());
+ }
+
+ @JsonCreator
+ public Event(
+ @JsonProperty("id") UUID id,
+ @JsonProperty("type") String type,
+ @JsonProperty("attributes") Map<String, Object> attributes,
+ @JsonProperty("attachments") Map<String, Object> attachments) {
if (type == null || type.isEmpty()) {
throw new IllegalArgumentException("Event 'type' must not be null or empty.");
}
this.id = id;
this.type = type;
this.attributes = attributes != null ? attributes : new HashMap<>();
+ this.attachments = attachments != null ? attachments : new HashMap<>();
}
Review Comment:
Event stores the provided `attachments` map instance directly. If callers pass an immutable map (e.g., `Map.of(...)`) or a shared map, runtime code that wraps attachments into `MemoryRef` (via `EventAttachmentUtils.storeEventAttachments`) will fail with `UnsupportedOperationException` or mutate caller-owned state. Copy `attributes`/`attachments` into new mutable maps in the constructor.
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -125,7 +149,16 @@ public static Event fromEvent(Event event) {
* @throws IOException if JSON parsing fails or the 'type' field is missing or empty
*/
public static Event fromJson(String json) throws IOException {
- return MAPPER.readValue(json, Event.class);
+ Event event = MAPPER.readValue(json, Event.class);
+ for (Map.Entry<String, Object> entry : event.getAttachments().entrySet()) {
+ Object attachment = entry.getValue();
+ if (attachment instanceof Map
+ && ((Map<?, ?>) attachment).containsKey(MemoryRef.MEMORY_TYPE_FIELD)
+ && ((Map<?, ?>) attachment).containsKey(MemoryRef.PATH_FIELD)) {
+ entry.setValue(MAPPER.convertValue(attachment, MemoryRef.class));
+ }
Review Comment:
`fromJson` converts any attachment map containing `memory_type` and `path` into a `MemoryRef`, even if the map has additional keys. That can unexpectedly reinterpret user-provided attachment objects. Make the conversion strict (only when the map has exactly those two keys), matching the Python side behavior.
##########
python/flink_agents/runtime/memory/event_attachment_utils.py:
##########
@@ -0,0 +1,90 @@
+################################################################################
+# 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.
+#################################################################################
+from __future__ import annotations
+
+import hashlib
+from typing import TYPE_CHECKING, Any
+
+from flink_agents.api.events.event import OutputEvent
+from flink_agents.api.memory_object import validate_memory_value
+from flink_agents.api.memory_reference import MemoryRef
+
+if TYPE_CHECKING:
+ from uuid import UUID
+
+ from flink_agents.api.events.event import Event
+ from flink_agents.api.runner_context import RunnerContext
+
+_ATTACHMENT_ROOT = "__event_attachments__"
+
+
+class EventAttachmentError(RuntimeError):
+ """Raised when an Event attachment cannot be stored or loaded."""
+
+
+def _hash_attachment_key(key: str) -> str:
+ return hashlib.sha256(key.encode("UTF-8")).hexdigest()
+
+
+def build_attachment_path(event_id: UUID, key: str) -> str:
+ """Build the canonical SensoryMemory path for one attachment."""
+ return f"{_ATTACHMENT_ROOT}.{event_id}.{_hash_attachment_key(key)}"
+
+
+def _attachment_context(event: Event, key: str, path: str | None = None) -> str:
+ suffix = f", path={path}" if path is not None else ""
+ return f"event_id={event.id}, event_type={event.type}, key={key}{suffix}"
+
+
+def store_event_attachments(event: Event, ctx: RunnerContext) -> None:
+ """Store concrete attachment values in SensoryMemory and replace them with refs."""
+ if not event.attachments:
+ return
+
+ if event.type == OutputEvent.EVENT_TYPE:
+ keys = ", ".join(sorted(event.attachments))
+ msg = f"Output events cannot carry attachments: {_attachment_context(event, keys)}"
+ raise EventAttachmentError(msg)
+
Review Comment:
Python `store_event_attachments` currently rejects attachments on `OutputEvent`, but this PR also adds attachments to cross-language OutputEvent snapshots and copies attachments in `OutputEvent.from_event(...)`. This mismatch will make `ctx.send_event(OutputEvent(..., attachments=...))` fail on Python while the Java runtime accepts it. Either allow OutputEvent attachments here or enforce the same restriction on both runtimes/APIs.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3944593524
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java:
##########
@@ -144,6 +145,52 @@ void testExecuteAgent() throws Exception {
}
}
+ @Test
+ void testJavaEventAttachmentsAreOffloadedAndResolvedBetweenActions() throws Exception {
+ InMemoryActionStateStore actionStateStore = new InMemoryActionStateStore(false);
+ AgentPlan agentPlan = TestAgent.getEventAttachmentAgentPlan();
+ try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new ActionExecutionOperatorFactory<>(agentPlan, true, actionStateStore),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class))) {
+ testHarness.open();
+ ActionExecutionOperator<Long, Object> operator =
+ (ActionExecutionOperator<Long, Object>) testHarness.getOperator();
+
+ List<Event> eventsAtSendBoundary = new ArrayList<>();
+ operator.getEventRouter()
+ .addEventListener(
+ (context, event) -> {
+ if (TestAgent.ATTACHMENT_EVENT_TYPE.equals(event.getType())) {
+ eventsAtSendBoundary.add(event);
+ }
+ });
+
+ testHarness.processElement(new StreamRecord<>(1L));
+ operator.waitInFlightEventsFinished();
+
+ assertThat(eventsAtSendBoundary).hasSize(1);
+ Event runtimeEvent = eventsAtSendBoundary.get(0);
+ Object reference = runtimeEvent.getAttachment(TestAgent.ATTACHMENT_KEY);
+ assertThat(reference).isInstanceOf(MemoryRef.class);
+ List<StreamRecord<Object>> recordOutput =
+ (List<StreamRecord<Object>>) testHarness.getRecordOutput();
+ assertThat(recordOutput).hasSize(1);
+ assertThat(recordOutput.get(0).getValue()).isEqualTo(Map.of("value", 1L));
+
+ ActionState actionState =
+ actionStateStore.get(
+ 1L,
+ 0L,
+ agentPlan.getActions().get("receiveEventAttachment"),
+ runtimeEvent);
+ assertThat(actionState).isNotNull();
+ assertThat(actionState.getTaskEvent().getAttachment(TestAgent.ATTACHMENT_KEY))
+ .isSameAs(reference);
Review Comment:
Good catch, thanks. Since the existing assertion already covers the intended behavior, I think this assertion can simply be removed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3887236812
##########
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 for pointing this out. Explicitly testing the behavior at the action boundaries is necessary here.
Added new test at [28da154e](https://github.com/apache/flink-agents/pull/950/changes/6019ef9fea507723983fcc56c13bb67a6940fc7c)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3837388709
##########
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:
Yes, this closes it. The runtime event is only read now, so the one the durable store holds keeps its refs.
One thing I'm unsure about: the copy is unconditional. `EventAttachmentUtils.java:97` copies before looking at anything, and every Java action goes through it (`JavaActionTask.java:61`), so an event with no attachments still pays for a full deep copy. `storeEventAttachments` bails out early on an empty map at `:51-53`, and the load side could do the same, since the only write is at `:111`, behind the `instanceof MemoryRef` check.
Then again, maybe the uniform copy is the point. The javadoc at `:95` calls it an action-owned copy, and that isolation also keeps an action from mutating the live event's attributes, which `ActionStateUtil.generateUUIDForEvent` (`:65-67`) hashes into the durable state key. Which way were you thinking?
Smaller one at `:41-42`: the serializer comes from a bare `new SerializerConfigImpl()`, so it has no registered serializers, while keyed state builds its own from the job config. A type registered through `pipeline.serialization-config` would apply when the event is persisted but not to this copy. Would building it from the operator's config and holding it as an instance field work? That drops the static `ThreadLocal` too.
nit: `resolvesAttachmentsWithoutMutatingDurableEvent` checks the subclass and the attachments but nothing about `attributes`, so a copy that dropped them would still pass. `assertThat(((InputEvent) invokedEvent).getInput()).isEqualTo(1L)` would cover it, if useful.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3743281042
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -125,7 +149,19 @@ public static Event fromEvent(Event event) {
* @throws IOException if JSON parsing fails or the 'type' field is missing or empty
*/
public static Event fromJson(String json) throws IOException {
- return MAPPER.readValue(json, Event.class);
+ Event event = MAPPER.readValue(json, Event.class);
+ for (Map.Entry<String, Object> entry : event.getAttachments().entrySet()) {
Review Comment:
Done，with this [commit](https://github.com/apache/flink-agents/pull/950/changes/600b90158d6645e04c0ec989cd5cf1d7f9bdeb2c)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on PR #950:
URL: https://github.com/apache/flink-agents/pull/950#issuecomment-5161461137
Hello @wenjin272, Could you take a look at this PR when you have time? Thank you very much
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3837858366
##########
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:
All three look right, thanks.
One loose end on my side: the isolation half of that question is not a concern. At the PR base (`3070ee21`) `JavaActionTask.java:75` passed the live event straight to the action, so skipping the copy when nothing needs resolving puts that case back where it was.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3837858366
##########
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:
All three look right, thanks.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3743271920
##########
python/flink_agents/api/events/event.py:
##########
@@ -72,11 +74,14 @@ 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.
+ attachments : Dict[str, Any]
+ Key-value data passed between actions through sensory memory.
"""
id: UUID = Field(default=None)
type: str
attributes: Dict[str, Any] = Field(default_factory=dict)
+ attachments: Dict[str, Any] = Field(default_factory=dict)
Review Comment:
Yes, this is a clear bug, and I have fixed it.
While working on the fix, I also noticed that Python generated event IDs from an MD5 hash of serialized event content. As a result, two events with identical attributes could receive the same ID, which could cause attachment values to overwrite each other because their MemoryRef paths are derived from the event ID.
I have also updated Python to generate event IDs with uuid.uuid4(), matching Java’s UUID.randomUUID() behavior. If Java and Python intentionally used different ID-generation strategies for historical reasons, please let me know.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3743361878
##########
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:
I forgot the `setAttachment` function in Java. It has been fixed now.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3836719754
##########
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:
Thanks for point that. I wasn’t familiar with `durable execution`, so I missed that resolving the attachments also modified the event stored in ActionState
In [ba79e7dd](https://github.com/apache/flink-agents/pull/950/changes/ba79e7dd57c538ff1e01c258ed1595ccfbdc54df), I changed Java attachment resolution to operate on an copy of the event instead of mutating the original instance. I also added a durable-execution test to verify that.
Could you please help check this is corret now? Thanks!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature][api][runtime] Support auto resolve memory reference for passing data across actions. [flink-agents]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on PR #950:
URL: https://github.com/apache/flink-agents/pull/950#issuecomment-5683066181
Hello @wenjin272, @weiqingy
Thanks for the reminder. I have resolved the conflicts and updated MemoryRef's Java API documentation from `Short-Term Memory` to `{@link MemoryObject}` at [7fc7cc8](https://github.com/apache/flink-agents/pull/950/commits/7fc7cc83b5a2ac9d2e61a3777089a7e72d1b29de).
Since this PR changes many files and is likely to conflict with other branches, I think add user documentation in another PR is better. If there are no other issues, I think this PR is ready to merge now.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org