[runtime] Preserve byte[] MemoryUpdate values across durable-execution serde (#865)

[weiqingy]

Linked issue: #865

Purpose of change

ActionStateSerde serializes MemoryUpdate.value (declared Object) via a plain, type-info-less Jackson ObjectMapper, so a byte[] is written as a base64 String and — with no type tag — deserialized back as String on crash/replay, silently writing the wrong type into memory.

This is a pre-existing, Java-native defect: a Java agent can set a byte[] memory value directly, so the bug predates admitting Python bytes into the memory contract (#846); #846 only makes it more likely to be hit.

The fix adds a targeted, field-scoped envelope (de)serializer bound to MemoryUpdate.value via a Jackson mixin. A byte[] is written as {"__flink_agents_bytes__":"<base64>"} and reversed back to byte[] on read; every other value type delegates to Jackson's stock untyped-Object (de)serialization and stays byte-identical on disk, so in-flight durable state remains replay-compatible.

The (de)serializer is bound per-property (mixin on value), not registered globally on byte[], so CallResult's statically-typed byte[] fields are untouched. The envelope carries only base64 text → byte[], so it adds no polymorphic-typing deserialization surface.

Scope: only a top-level byte[] value is preserved. A byte[] nested inside a List/Map value goes through the default serializers and is out of scope, documented in the code.

Tests

Added round-trip regressions in ActionStateSerdeTest (the first mirrors the issue repro exactly):

• byte[] on a short-term memory update round-trips as byte[] (issue repro; assertArrayEquals)
• byte[] on a sensory memory update
• empty byte[0]
• a String value still round-trips as String (envelope doesn't disturb the common case)
• a Map value still round-trips as a Map (not misread as an envelope; on-disk shape unchanged)
• reserved-key residual: a user Map that is exactly {"__flink_agents_bytes__":"<base64>"} recovers as byte[] (documented residual), while any second key keeps it a Map

mvn test -Dtest=ActionStateSerdeTest -pl runtime → 16 tests, 0 failures. mvn test -pl runtime → 396 tests, 0 failures.

API

No public API change. MemoryUpdate, ActionState, and ActionStateSerde.serialize/deserialize signatures are unchanged; only the on-disk encoding of a byte[] value changes (it was previously broken).

Documentation

• doc-needed
• doc-not-needed
• doc-included

[weiqingy]

Closing this in favor of #874, which supersedes it.

Per the discussion on #872, rather than land this narrow top-level byte[] envelope and then generalize it, we fold everything into one binary change: #874 switches MemoryUpdate.value to a Flink-native (Kryo) binary payload inside a versioned {serde, version, payload} envelope, which covers the top-level byte[] case this PR fixed plus nested byte[], Long narrowing, and user POJOs — and retires the __flink_agents_bytes__ envelope. Nothing here is persisted in that format, so there is no migration concern.