[Bug] byte[] memory values corrupted to base64 String on the durable-execution replay path

[wenjin272]

Search before asking

• I searched in the issues and found nothing similar.

Description

The durable-execution path (ActionStateStore, Kafka/Fluss) serializes MemoryUpdate.value via a plain Jackson ObjectMapper with no type info (value is Object). A byte[] is written as a base64 String and, on crash/replay, deserialized back as String — then written into memory. The consumer silently gets str/String instead of the original bytes.

How to reproduce

Add to ActionStateSerdeTest (or run equivalently):

ActionState state = new ActionState(null, null, null, null, null, false);
state.addShortTermMemoryUpdate(new MemoryUpdate("stm.path", "hello".getBytes()));

ActionState out = ActionStateSerde.deserialize(ActionStateSerde.serialize(state));
Object value = out.getShortTermMemoryUpdates().get(0).getValue();

// Expected: byte[] equal to "hello".getBytes()
// Actual:   value is a String "aGVsbG8=" (base64), type lost

Round-tripping through ActionStateSerde turns the byte[] into a base64
String, so on crash/replay the wrong type is written back into memory.

Version and environment

0.1~0.3

Are you willing to submit a PR?

• I'm willing to submit a PR!

[weiqingy]

@wenjin272 I'd like to take this one — it's the direct tail of #846.