minxhe opened a new pull request, #29575: URL: https://github.com/apache/flink/pull/29575
## What is the purpose of the change `org.apache.flink.api.common.state.v2.MapState` specifies that `asyncPut(k, null)` and null-valued entries in `asyncPutAll(map)` remove the corresponding keys, an empty map is a no-op, and a `null` map argument is forbidden. `MapStateAdaptor` (which backs State V2 `MapState` on V1 state backends such as `HashMapStateBackend` and `EmbeddedRocksDBStateBackend` via `AsyncKeyedStateBackendAdaptor`) delegated `asyncPut` and `asyncPutAll` directly to V1 `InternalMapState#put` and `#putAll`: - On `HeapMapState`, `put(k, null)` and `putAll` with `null` entry values stored a `null` value in the map instead of removing the key (`asyncContains(k)` remained `true` and `asyncEntries()` still emitted the key). - On `RocksDBMapState`, `putAll` serialized `null` values as null markers instead of deleting the entries (`asyncContains(k)` remained `true`). Jira: https://issues.apache.org/jira/browse/FLINK-41095 ## Brief change log - In `MapStateAdaptor#asyncPut`, call `delegatedState.remove(key)` when `value == null`. - In `MapStateAdaptor#asyncPutAll`, check `checkNotNull(map, "Map must not be null")`, short-circuit on `map.isEmpty()`, and when the map contains `null` values, remove those keys via `delegatedState.remove(key)` while forwarding any non-null entries to `delegatedState.putAll`. - Add conformance and exception-wrapping tests in `AsyncKeyedStateBackendAdaptorTest` and `EmbeddedRocksDBStateBackendTest`. ## Verifying this change This change added tests and can be verified as follows: - `AsyncKeyedStateBackendAdaptorTest#testMapStateNullRemovals` and `#testAsyncMapStateMutationExceptions` fail on unmodified `master` (`8d01380ff5bf9892462ef9808b1b56fd9723f54d`) because `asyncContains` remains `true` after `asyncPutAll` / `asyncPut` with `null` values and `asyncPutAll(Map.of())` invokes the delegate. - Both tests pass with the fix, along with `EmbeddedRocksDBStateBackendTest#testAsyncMapStateNullRemovals` (4 parameterized invocations). ```sh ./mvnw -B -ntp -pl flink-runtime -Dflink.forkCountUnitTest=1 \ -Dtest=AsyncKeyedStateBackendAdaptorTest test # 7 run, 0 failures ./mvnw -B -ntp -pl flink-state-backends/flink-statebackend-rocksdb -Dflink.forkCountUnitTest=1 \ -Dtest='EmbeddedRocksDBStateBackendTest#testAsyncMapStateNullRemovals' test # 4 run, 0 failures ./mvnw -B -ntp -pl flink-runtime,flink-state-backends/flink-statebackend-rocksdb \ spotless:check checkstyle:check # 0 violations ``` ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): yes (`MapStateAdaptor#asyncPut` and `#asyncPutAll` on V1-adapted async state; maps without `null` values take a fast path to `delegatedState.putAll(map)`) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: GitHub Copilot CLI 1.0.99-2 -- 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]
