[PR] [runtime][java] Reject memory writes below a value field before mutating state [flink-agents]

["sangkyoonnam (via GitHub)" <gi...@apache.org>]

sangkyoonnam opened a new pull request, #1188:
URL: https://github.com/apache/flink-agents/pull/1188

   Linked issue: #1187
   
   ### Purpose of change
   
   #### User-visible outcome
   
   A `set` or `newObject` whose path runs below a field that holds a value now throws `IllegalArgumentException` and leaves nothing behind. The message names the absolute requested path and offending ancestor; with `o` an object and `o.a` a value, `get("o").set("a.b", v)` reports `Cannot write field 'o.a.b': 'o.a' exists but is not an object.`. Previously these calls threw `UnsupportedOperationException`; deeper paths could leave orphans, while `newObject("a.b")` also recorded an update that made replay fail after restore.
   
   A path with an empty component (`"a."`, `".a"`, `"a..b"`, `""`) is rejected the same way, with `... path has an empty component.`. Before, such paths threw `NullPointerException` or `UnsupportedOperationException`, stored a key the parent could not list, or were accepted: `set("a..b", v)` with `a` absent stored a readable value, and root `newObject("")` recorded an update and registered the root as its own field. Those two are now rejected on purpose. Root `set("")` already threw, after registering that field.
   
   #### Intent
   
   Validate the whole write path before any mutation, so a rejected write can't corrupt keyed state or the updates recovery replays.
   
   #### Runtime flow
   
   `set` and `newObject` call `checkWritePath` before `fillParents`. It splits with `split("\\.", -1)` to keep trailing empty components, rejects any empty component, then walks every proper prefix and throws if the item stored there exists and is not an OBJECT.
   
   #### Key decisions
   
   - `IllegalArgumentException`, as for the existing overwrite conflicts.
   - `overwrite=true` on `newObject` still applies to the target field only; replacing a value ancestor would drop its value silently.
   - Empty components are rejected rather than given a meaning.
   - The check costs one store `get` per proper prefix, none for a top-level write.
   - No change to `MemoryUpdateReplayer`. An `ActionState` written before this change can still carry a rejected update; replaying it now fails with this message. Skipping it on replay is a separate decision I'll leave out unless you'd rather handle it here.
   
   ### Behavioral Semantics
   
   #### Interaction decisions
   
   | Memory state | Call | Before | After |
   |---|---|---|---|
   | `a` is a value | `set("a.b", v)`, `get("a").set("b", v)` | `UnsupportedOperationException`, no mutation | `IllegalArgumentException`, no mutation |
   | `a` is a value | `set("a.b.c", v)`, `newObject("a.b.c")` | same, orphan `a.b` persisted | `IllegalArgumentException`, no mutation |
   | `a` is a value | `newObject("a.b")`, either `overwrite` | same, orphan `a.b` and update recorded | `IllegalArgumentException`, no mutation, nothing recorded |
   | any | empty path component | NPE, UOE, unlistable key, or accepted (two cases above) | `IllegalArgumentException`, no mutation |
   
   #### Behavioral contracts
   
   1. A write below a value field, or with an empty path component, throws `IllegalArgumentException` naming the absolute requested path.
   2. The rejected write leaves keyed state untouched: no intermediate node, and the value and its parent's field list as they were.
   3. The rejected write records no `MemoryUpdate`, so replaying a completed action that caught it reproduces only the writes that succeeded.
   4. Writes whose proper prefixes are all objects or absent, with no empty components, behave as before.
   
   #### Failure behavior
   
   The exception is thrown before `fillParents`, so nothing needs rolling back.
   
   ### Tests
   
   #### Contracts to tests
   
   | Contract | Tests |
   |---|---|
   | 1, 2 | the four `MemoryObjectTest.test*IsRejectedWithoutSideEffects` tests |
   | 2 across restore | `MemoryWriteBelowValueRecoveryTest.rejectedWriteBelowValueLeavesNoOrphanAcrossRestore` (both backends; an action reads value, fields and `isExist` through `MemoryObject` before and after restore) |
   | 3 | `MemoryUpdateReplayerTest.testReplayOfActionThatCaughtRejectedNewObjectUnderValue`; `MemoryWriteBelowValueRecoveryTest.replayOfActionThatCaughtRejectedNewObjectCompletesRecovery` (both backends, redelivery after restore) |
   | 4 | existing tests in those classes and `ActionExecutionOperatorTest` |
   
   #### Coverage and what was not verified
   
   With `MemoryObjectImpl` reverted to `main`, all 9 new runs fail (7 methods, 2 parameterized; modes in the details below). With the change: `mvn -pl api test` 523 run, 0 failures, 12 skipped; `mvn -pl runtime test` 1151 run, 0 failures, 1 skipped.
   
   Not verified: a job with a restart strategy (I ran one redelivery after restore; replay is deterministic, so each restart should fail the same way); the serial replay path at `ActionExecutionOperator.java:772`; Kafka and Fluss action state stores; JDK 11 and 17 test runs; Python past a Pemja probe.
   
   Separate defect, left alone here: `set` still accepts and records a `MemoryObject` value (`MemoryObjectImpl.java:114`) although the javadoc lists that as an error.
   
   <details>
   <summary>Supporting evidence</summary>
   
   - On `main`, the new tests fail with a wrong exception type, an extra recorded update, and `UnsupportedOperationException` at `MemoryObjectImpl.newObject:148` under `MemoryUpdateReplayer.replay:51` in the recovery test on both backends.
   - The orphan `a.b` was observed in keyed state on both backends and survived `snapshot` + `initializeState`. The recorded update was observed in the `InMemoryActionStateStore` retained across the restore, where replay reads it from.
   - `tools/lint.sh -c` passes on JDK 11.
   - `FlinkMemoryObject` delegates to the Java object and needs no change; a local Pemja probe showed the Java exception wrapped in `MemoryObjectError`. The `LocalMemoryObject` test double differs on this sequence and is left alone here.
   </details>
   
   ### API
   
   #### Compatibility impact
   
   No signature changes. These writes are now intentionally rejected before mutation with `IllegalArgumentException`, including formerly accepted malformed paths. The `MemoryObject` javadoc names both cases on `set` and both `newObject` overloads. Nothing in the repository catches the old types or writes such paths.
   
   ### Documentation
   
   <!-- Do not remove this section. Check the proper box only. -->
   
   - [ ] `doc-needed` <!-- Your PR changes impact docs -->
   - [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
   - [ ] `doc-included` <!-- Your PR already contains the necessary documentation updates -->
   
   ### Was this patch authored or co-authored using generative AI tooling?
   
   <!-- Do not remove this section. Check the proper box only. -->
   
   - [x] Yes
   - [ ] No
   
   Generated-by: Claude Code 2.1.284 (Claude Fable 5.1)
   


-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [runtime][java] Reject memory writes below a value field before mutating state [flink-agents]

["wenjin272 (via GitHub)" <gi...@apache.org>]

wenjin272 merged PR #1188:
URL: https://github.com/apache/flink-agents/pull/1188


-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org