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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
