[api][runtime] Replay newObject memory updates as object creations during durable recovery

[purushah]

Linked issue: #1029

Purpose of change

With durable execution enabled, MemoryObjectImpl.newObject() records itself in the ActionState as MemoryUpdate(path, null) — indistinguishable from a user calling set(path, null). The recovery replay path applies every recorded update via set(), so replaying a completed action that created a nested object either throws IllegalArgumentException: Cannot overwrite object with value against an object restored from the checkpoint (a permanent recovery crash loop — reproduced on a local mini-cluster job with a Kafka action-state store, where the job burned all restart attempts and died), or materializes the object as a null value leaf, after which replaying any child write throws UnsupportedOperationException (and the memory tree is silently corrupted).

This PR adds an objectCreation discriminator to MemoryUpdate, records it from newObject(), and replays object-creation updates via newObject(path, true) in a new shared MemoryUpdateReplayer used by the operator's short-term and sensory replay loops.

Tests

• New MemoryUpdateReplayerTest (4 cases): replay into empty state restores a real nested object with children; replay over an already existing object (the crash-loop case) succeeds; newObject(path, overwrite=true) over a value leaf replays correctly; a user's genuine set(path, null) still replays as a null value, not an object.
• Two operator-level recovery tests in ActionExecutionOperatorTest (testReplayReappliesNewObjectMemoryUpdatesIntoEmptyState, testReplayReappliesNewObjectMemoryUpdatesOverRestoredState) run both scenarios end to end through KeyedOneInputStreamOperatorTestHarness with an ActionStateStore, including checkpoint snapshot/restore.
• ActionStateSerdeTest extended with a round-trip assertion for the new field.
• All of the above verified red before the fix (exact exceptions above) and green after; the affected runtime suites (130 tests, including all of ActionExecutionOperatorTest and ShortTermMemoryTTLIntegrationTest) pass.
• Additionally verified with a real local Flink job (mini-cluster, checkpointing, Kafka action-state store, injected one-time failure): on current main the job crash-loops and dies with Cannot overwrite object with value: user; with this patch the identical job recovers and completes with correct memory state. Logs attached to the linked issue.

API

MemoryUpdate gains a three-arg constructor and isObjectCreation(); the existing two-arg constructor is kept (source- and binary-compatible). Compatibility notes:

• Old durable journals read by new code: the missing objectCreation field defaults to false, so pre-existing records replay exactly as before.
• New journals read by old code: rejected — consistent with ActionStateSerde's documented recovery-compat contract ("the durable-execution journal ... never survives a code or Flink upgrade").
• Flink checkpoints/savepoints: untouched (no change to MemoryItem or state schemas).
• Behavior change: MemoryUpdate.equals/hashCode/toString now include the flag, so a recorded newObject update no longer compares equal to set(path, null) — which is the semantic distinction this fix introduces.
• Memory-event JSON schemas and the Python surface are unchanged.

Documentation

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

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

Generated-by: Claude Code 2.1.234 (Claude Fable 5)

🤖 Generated with Claude Code

[wenjin272]

Hi @joeyutong, could you help review this pr?

[joeyutong]

Thanks for addressing all three comments. I re-reviewed 4edd5f6 and don’t see any further issues with the change itself. The only remaining item is to rebase and resolve the conflicts with current main. When resolving ActionExecutionOperator, please keep notifyActionReused(actionTask) after both memory replays. Once the conflicts are resolved, this looks good to me.

[purushah]

Rebased onto current main (0e36efd). The ActionExecutionOperator conflict was resolved exactly as requested — notifyActionReused(actionTask) sits after both MemoryUpdateReplayer.replay(...) calls (short-term, then sensory). The test-file conflict kept both your new testActionStateStoreReplayRecordsReusedExecutions and the two recovery tests from this PR. Verified locally on the rebased tree: spotless clean, 79/79 affected tests green including the full ActionExecutionOperatorTest suite (47 tests, trace-replay test included).