[runtime] Binary serde for MemoryUpdate values via versioned Kryo envelope#874
Merged
Merged
Conversation
github-actions
Bot
added
doc-not-needed
github-actions
Bot
added
doc-not-needed
…elope
MemoryUpdate.value is declared Object and was persisted by ActionStateSerde
through a type-info-less Jackson mapper, so on durable-execution replay any
value that bare JSON cannot reconstruct degraded silently: byte[] to base64
String, int-range Long to Integer, user POJO to LinkedHashMap, and the same
for those types nested in a List or Map. The recovered wrong type then flowed
into agent memory unchecked.
Bind a field-scoped Jackson (de)serializer to MemoryUpdate.value only that runs
the Flink-native serializer (Kryo for a generic Object) and stores the bytes
base64-encoded inside a self-describing versioned envelope embedded in the
single ActionState JSON document: {serde, version, payload}. One mechanism
covers every Kryo-serializable type. Field-scoping keeps CallResult's
statically-typed byte[] payloads untouched. The Kryo serializer is built once
and duplicated per thread via a ThreadLocal for thread-safety. An unknown
envelope version is rejected rather than silently mis-decoded.
Generalizes the top-level byte[] fix and supersedes the narrow byte[] envelope.
Safe because the durable-execution journal is checkpoint-scoped, always written
and read by the same code and Flink version, so Kryo cross-version binary
incompatibility is never exercised.
Closes apache#872
weiqingy
force-pushed
the
872-memoryupdate-binary-serde
branch
from
c02042f to
f0a64a0
Compare
wenjin272
reviewed
Jul 8, 2026
| @Override | ||
| public Object deserialize(JsonParser p, DeserializationContext ctxt) throws IOException { | ||
| JsonNode node = ctxt.readTree(p); | ||
| if (isEnvelope(node)) { |
Contributor
There was a problem hiding this comment.
IIUC, the isEnvelope flag is used to maintain backward compatibility with legacy MemoryUpdate instances that were serialized using JSON.
Is it essential to maintain backward compatibility in this case?
- For new jobs, both serialization and deserialization should be performed using Kryo, with
isEnvelopeset to true. - For old jobs, both serialization and deserialization should be performed using Jackson.
- The
isEnvelopecan only become false after a job upgrade and restart. Since flink agents are still in beta, I consider this incompatibility acceptable.
Collaborator
Author
There was a problem hiding this comment.
Thanks for the review, @wenjin272. I've updated the PR accordingly.
…allback Within a job every non-null MemoryUpdate value is written as a Kryo envelope, and the durable journal is checkpoint-scoped so it never outlives the code that wrote it. A non-envelope node can therefore only come from an unsupported cross-version restore; silently binding it via the stock untyped-Object deserializer would reintroduce the type loss this change fixes (e.g. byte[] returning as a base64 String). Reject it with a clear error instead. Null values are unaffected: Jackson routes a JSON null through getNullValue(), not the custom deserializer.
This was referenced Jul 8, 2026
Closed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Linked issue: closes #865, closes #872
Purpose of change
MemoryUpdate.valueis declaredObject, andActionStateSerdepersisted it through a JacksonObjectMapperthat carries no type information for that field. On durable-execution replay the value was rebuilt by Jackson's stock untyped deserializer, so any runtime type that bare JSON cannot reconstruct degraded silently:byte[]to base64String, an int-rangeLongtoInteger, a user POJO toLinkedHashMap, and the same losses for those types nested inside aListorMap. The consumer performs no downcast, so the wrong type flowed into agent memory after any crash or restore. This generalizes #865 / #871, which fixed only a top-levelbyte[].The fix binds a field-scoped Jackson (de)serializer to
MemoryUpdate.valueonly. It runs the Flink-native serializer — Kryo, for a genericObject— and stores the bytes base64-encoded inside a self-describing versioned envelope embedded in the singleActionStateJSON document:{ "serde": "kryo", "version": 1, "payload": "<base64>" }. One mechanism covers every Kryo-serializable type instead of a per-type special case.Key points:
MemoryUpdate.valuevia a mixin, never a globalbyte[]serializer, soCallResult's statically-typedbyte[]payloads keep round-tripping unchanged.KryoSerializeris not thread-safe, so one immutable template is built once and each thread serializes through its ownduplicate()via aThreadLocal.versionis rejected rather than silently mis-decoded.versionmarker, Kryo pinned within the major release) is documented on the serde.Eventfield stays JSON, preserving the cross-language and event-log contract.This supersedes #871 (the narrow
__flink_agents_bytes__byte[] envelope), which is unmerged, so no state is persisted in that format.Tests
Extends
ActionStateSerdeTestwith 5 round-trip tests: top-levelbyte[], an int-rangeLong, a nestedMap/Listgraph holding both abyte[]and aLong, a custom POJO (recovered as its class, not aLinkedHashMap), and an unknown-version envelope (rejected). The 10 existing tests are unchanged.mvn -pl runtime -am test -Dtest=ActionStateSerdeTestpasses 15 tests under the default Flink 2.3.0, which is also the runtime linkage check for the Kryo construction path.mvn -pl runtime spotless:checkis clean.API
No public API change. All new types are package-private members of
ActionStateSerde. No new dependency, and nodist/change — Kryo is provided by the cluster Flink.Documentation
doc-neededdoc-not-neededdoc-included