purushah opened a new issue, #1029:
URL: https://github.com/apache/flink-agents/issues/1029
### Summary
With durable execution enabled (a Kafka/Fluss `ActionStateStore`), replaying
a completed action that called `MemoryObject.newObject()` either **crash-loops
recovery permanently** or **silently corrupts the short-term memory tree**,
depending on whether the object already exists in the restored checkpoint.
### Root cause
`MemoryObjectImpl.newObject()` records its effect in the action state as a
plain null-value write:
```java
// runtime/.../memory/MemoryObjectImpl.java:141
memoryUpdates.add(new MemoryUpdate(absPath, null));
```
`MemoryUpdate` carries no object-vs-value discriminator, so this record is
byte-for-byte indistinguishable from a user calling `set(path, null)`. On
recovery, the replay path applies **every** recorded update via `set()`:
```java
// runtime/.../operator/ActionExecutionOperator.java:420-431
for (MemoryUpdate memoryUpdate : actionState.getShortTermMemoryUpdates()) {
actionTask.getRunnerContext().getShortTermMemory()
.set(memoryUpdate.getPath(), memoryUpdate.getValue());
}
```
The type information is already lost when the update is recorded, so this is
not fixable at the serde layer (the versioned Kryo envelope from #872 correctly
preserves value *types*, but not the update *kind*).
### Failure modes
Let an action do the documented nested-memory pattern — `newObject("user");
set("user.score", ...)` — complete, get recorded in the ActionState store, and
then have its input re-delivered after a failover:
1. **`user` exists in the restored checkpoint** (e.g. an earlier input for
the same key created it; the replayed action used get-or-create
`newObject("user")`): replay executes `set("user", null)` →
`IllegalArgumentException: Cannot overwrite object with value: user`
(`MemoryObjectImpl.java:110-112`). The same exception recurs on **every**
restart attempt → permanent recovery crash loop until restart attempts are
exhausted and the job dies.
2. **`user` is not in the restored checkpoint**: replay materializes it as a
VALUE leaf with value `null`. Replaying the child write `set("user.score",
...)` then calls `getSubKeys().add(...)` on the VALUE item's immutable
`Collections.emptySet()` → `UnsupportedOperationException`. Even without a
child write, the object has silently become a null leaf.
3. A user's legitimate `set(path, null)` and a `newObject(path)` are
indistinguishable in the persisted `ActionState` — `memory_events.md` already
acknowledges this ambiguity for *observability* events ("consumers cannot
distinguish an object marker from a real null value"), but the same lossy
record is used for **state reconstruction**.
### Reproduction 1 — real local job (mini-cluster + Kafka action state)
Job: rate-limited source of `"alice,<n>"` records → agent that does
`newObject("user")` (get-or-create) + `set("user.total", total)` → a map that
throws once, on the second record after a completed checkpoint (guaranteeing
one record's action completed after the checkpoint recovery restores from).
Checkpointing 2.5s, fixed-delay restart strategy (3 attempts),
`actionStateStoreBackend=kafka`.
Observed on current `main` (3070ee21):
```
[FailOnceMap] checkpoint 1 completed
alice: total=6 (+3) <- record 3's action completes AFTER checkpoint
1
<- record 4 triggers the injected failover
Caused by: ...ActionTaskExecutionException: Failed to execute action task
Caused by: java.lang.IllegalArgumentException: Cannot overwrite object with
value: user
at ...memory.MemoryObjectImpl.set(MemoryObjectImpl.java:111)
at
...operator.ActionExecutionOperator.processActionTaskForKey(ActionExecutionOperator.java:424)
...
Recovery is suppressed by
FixedDelayRestartBackoffTimeStrategy(maxNumberRestartAttempts=3,
backoffTimeMS=2000)
Exception in thread "main"
org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
```
Every restart attempt failed identically; the job never recovered. Note the
exposure condition: the guarded idiom `if (!isExist("user")) newObject("user")`
only records the creation on the *first* record per key, so it dodges the bug
unless that first record is in the replayed window; the equally valid
unconditional get-or-create `newObject("user")` (a no-op when the object
exists) hits it on any post-checkpoint record.
### Reproduction 2 — standalone operator-level repro (no Kafka needed)
Self-contained `main()` against the real `ActionExecutionOperator` via
`KeyedOneInputStreamOperatorTestHarness` + `InMemoryActionStateStore` (the same
store the existing replay tests use), covering both failure modes:
```
=== Scenario 1: recovery into empty keyed state ===
recovery FAILED during memory-update replay:
Caused by: java.lang.UnsupportedOperationException
at
java.base/java.util.AbstractCollection.add(AbstractCollection.java:251)
at ...memory.MemoryObjectImpl.set(MemoryObjectImpl.java:106)
at
...operator.ActionExecutionOperator.processActionTaskForKey(ActionExecutionOperator.java:424)
=== Scenario 2: recovery over restored state (crash loop) ===
recovery FAILED during memory-update replay (would repeat on every restart
=> permanent crash loop):
Caused by: java.lang.IllegalArgumentException: Cannot overwrite object with
value: user
at ...memory.MemoryObjectImpl.set(MemoryObjectImpl.java:111)
at
...operator.ActionExecutionOperator.processActionTaskForKey(ActionExecutionOperator.java:424)
RESULT: 2 of 2 recovery scenarios FAILED (bug reproduced).
```
I can attach the full repro source and job logs.
### Proposed fix
Add an `objectCreation` discriminator to `MemoryUpdate` (additive: old 2-arg
constructor kept; Jackson defaults the missing field to `false` for records
written before the field existed, preserving their replay behavior), record it
from `newObject()`, and replay object-creation updates via `newObject(path,
true)` instead of `set(path, null)`. Verified: both reproductions above pass
with the fix, and the affected runtime suites (130 tests incl. all of
`ActionExecutionOperatorTest` and the TTL integration tests) stay green. PR to
follow.
### Environment
- current `main` @ 3070ee21
- Flink 2.3.0, JDK 17, macOS (local mini-cluster) and operator test harness
- Kafka action state store (apache/kafka:3.9.0) for the job-level repro
Generated-by: Claude Code 2.1.234 (Claude Fable 5)
--
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]
