purushah opened a new pull request, #1030:
URL: https://github.com/apache/flink-agents/pull/1030
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`
- [x] `doc-not-needed`
- [ ] `doc-included`
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
Generated-by: Claude Code 2.1.234 (Claude Fable 5)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
