weiqingy opened a new pull request, #874:
URL: https://github.com/apache/flink-agents/pull/874
Linked issue: #872
### Purpose of change
`MemoryUpdate.value` is declared `Object`, and `ActionStateSerde` persisted
it through a Jackson `ObjectMapper` that 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 base64 `String`, an int-range `Long` to
`Integer`, a user POJO to `LinkedHashMap`, and the same losses for those types
nested inside a `List` or `Map`. 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-level `byte[]`.
The fix binds a field-scoped Jackson (de)serializer to `MemoryUpdate.value`
only. It 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": "kryo",
"version": 1, "payload": "<base64>" }`. One mechanism covers every
Kryo-serializable type instead of a per-type special case.
Key points:
- Field-scoping is deliberate: the (de)serializer binds to
`MemoryUpdate.value` via a mixin, never a global `byte[]` serializer, so
`CallResult`'s statically-typed `byte[]` payloads keep round-tripping unchanged.
- Thread-safety: `KryoSerializer` is not thread-safe, so one immutable
template is built once and each thread serializes through its own `duplicate()`
via a `ThreadLocal`.
- An unknown envelope `version` is rejected rather than silently mis-decoded.
- Safe binary format: the durable-execution journal is checkpoint-scoped
(pruned on checkpoint completion, rebuilt on restore), so state is always
written and read by the same code and the same Flink version. It never crosses
a code or Flink upgrade, so Kryo's cross-version binary incompatibility and
schema-evolution brittleness are never exercised. The recovery-compat contract
(envelope shape, `version` marker, Kryo pinned within the major release) is
documented on the serde.
- The `Event` field 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 `ActionStateSerdeTest` with 5 round-trip tests: top-level `byte[]`,
an int-range `Long`, a nested `Map` / `List` graph holding both a `byte[]` and
a `Long`, a custom POJO (recovered as its class, not a `LinkedHashMap`), and an
unknown-version envelope (rejected). The 10 existing tests are unchanged.
`mvn -pl runtime -am test -Dtest=ActionStateSerdeTest` passes 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:check` is clean.
### API
No public API change. All new types are package-private members of
`ActionStateSerde`. No new dependency, and no `dist/` change — Kryo is provided
by the cluster Flink.
### Documentation
- [ ] `doc-needed`
- [x] `doc-not-needed`
- [ ] `doc-included`
--
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]
