You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "1996fanrui (via GitHub)" <gi...@apache.org> on 2025/07/25 07:47:05 UTC
[PR] [FLINK-38137][state] RocksDB and ForSt state backend are compatible with the case where the user value of the map state is null [flink]
1996fanrui opened a new pull request, #26831:
URL: https://github.com/apache/flink/pull/26831
## What is the purpose of the change
Currently, hashmap, RocksDB and ForSt are compatible with the case where the user value of the map state is null (Including write, read, checkpoint and restore), but it has an asymmetry that causing a NPE issue.
The core issue lies in the current implementation of serializeValueNullSensitive. It unconditionally calls valueSerializer.serialize(value, out) even when the value is null. This design has a significant flaw: the behavior of the state backend becomes dependent on the specific implementation of the TypeSerializer. If a serializer is not designed to handle null inputs, it will throw a NullPointerException, causing unexpected failures.
The most compelling argument for the change comes from the corresponding deserialization logic found in RocksDBMapState [https://github.com/apache/flink/blob/bf1cd860617f7b51ac91516814c0e931e5bba241/flin[…]c/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java](https://github.com/apache/flink/blob/bf1cd860617f7b51ac91516814c0e931e5bba241/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java#L413)
During deserialization, the code first reads a boolean flag to check if the stored value is null. If it is, the method immediately returns null and does not attempt to deserialize any further data from the input stream.
This reveals a critical asymmetry:
1. Serialization: Writes a isNull flag and then always attempts to serialize the value object, even if it's null.
2. Deserialization: Reads the isNull flag and, if true, never attempts to deserialize the value object.
The data written by serializer.serialize(null, ...) is therefore completely redundant and ignored during deserialization.
NPE Risk: The behavior becomes dependent on the TypeSerializer implementation. Serializers not designed to handle null inputs will throw NullPointerException.
Unfortunately, many core serializers in Flink, such as `org.apache.flink.api.common.typeutils.base.IntSerializer` and `org.apache.flink.api.common.typeutils.base.BooleanSerializer`, are designed to be null-unsafe and will throw a {@link NullPointerException} if they encounter a null value.
## Brief change log
- [FLINK-38137][state] RocksDB and ForSt state backend are compatible with the case where the user value of the map state is null
- Don't serialize user value when it's null
- The user value is not used during deserialization, so it doesn't introduce any incompatibility problem
- [FLINK-38137][state] Disable testMapStateWithNullValue for change log state backend temporarily
- Change log state backend is not compatible with the case where the user value of the map state is null
- Code path: https://github.com/apache/flink/blob/31785e076c86d0a44e3f4a17f44a04908a2d3eb4/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogMapState.java#L224
## Verifying this change
This change added tests and can be verified as follows:
- Added `StateBackendTestBase#testMapStateWithNullValue`, it checks the put and read map state when user value is null
- Added `MapStateNullValueCheckpointingITCase`, it checks whether checkpoint and restore work as expected when the user value of MapState is null.
## 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): no
- 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? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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] [FLINK-38137][state] RocksDB and ForSt state backend are compatible with the case where the user value of the map state is null [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #26831:
URL: https://github.com/apache/flink/pull/26831#issuecomment-3126424925
Thanks both for the review, merging
--
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] [FLINK-38137][state] RocksDB and ForSt state backend are compatible with the case where the user value of the map state is null [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #26831:
URL: https://github.com/apache/flink/pull/26831#discussion_r2230455833
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java:
##########
@@ -176,7 +176,10 @@ <T> byte[] serializeValueNullSensitive(T value, TypeSerializer<T> serializer)
throws IOException {
dataOutputView.clear();
dataOutputView.writeBoolean(value == null);
- return serializeValueInternal(value, serializer);
+ if (value != null) {
+ serializer.serialize(value, dataOutputView);
+ }
+ return dataOutputView.getCopyOfBuffer();
Review Comment:
Currently, hashmap, RocksDB and ForSt are compatible with the case where the user value of the map state is null (Including write, read, checkpoint and restore), but it has an asymmetry that causing a NPE issue. Here is core change, and ForSt is same. （Get more details from PR description.）
Unfortunately, change log state backend is not compatible with the case where the user value of the map state is null [1]. It could be improved to be compatible with this case, but it will break the serialization format and introduce some migration issues. I disabled `testMapStateWithNullValue` for change log state backend to avoid the CI is broken in this PR.
Hey @Zakelly , I hope to hear your opinion on how to move forward with the change log state backend?
If it is easier to solve, I would prefer to fix all problems of rocksdb, ForSt and change log together in the current PR. But if it is not, it might make more sense to fix the problems with rocksdb and ForSt serializers first.
WDYT?
[1] https://github.com/apache/flink/blob/31785e076c86d0a44e3f4a17f44a04908a2d3eb4/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogMapState.java#L224
--
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] [FLINK-38137][state] RocksDB and ForSt state backend are compatible with the case where the user value of the map state is null [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #26831:
URL: https://github.com/apache/flink/pull/26831#discussion_r2233898522
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java:
##########
@@ -176,7 +176,10 @@ <T> byte[] serializeValueNullSensitive(T value, TypeSerializer<T> serializer)
throws IOException {
dataOutputView.clear();
dataOutputView.writeBoolean(value == null);
- return serializeValueInternal(value, serializer);
+ if (value != null) {
+ serializer.serialize(value, dataOutputView);
+ }
+ return dataOutputView.getCopyOfBuffer();
Review Comment:
Thanks for the comment, I have created the https://issues.apache.org/jira/browse/FLINK-38144 to track it.
--
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] [FLINK-38137][state] RocksDB and ForSt state backend are compatible with the case where the user value of the map state is null [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26831:
URL: https://github.com/apache/flink/pull/26831#issuecomment-3116792848
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0ee438370e6c1517ef9d127c484abd6ae680fa6d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0ee438370e6c1517ef9d127c484abd6ae680fa6d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0ee438370e6c1517ef9d127c484abd6ae680fa6d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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] [FLINK-38137][state] RocksDB and ForSt state backend are compatible with the case where the user value of the map state is null [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #26831:
URL: https://github.com/apache/flink/pull/26831#discussion_r2232836074
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java:
##########
@@ -176,7 +176,10 @@ <T> byte[] serializeValueNullSensitive(T value, TypeSerializer<T> serializer)
throws IOException {
dataOutputView.clear();
dataOutputView.writeBoolean(value == null);
- return serializeValueInternal(value, serializer);
+ if (value != null) {
+ serializer.serialize(value, dataOutputView);
+ }
+ return dataOutputView.getCopyOfBuffer();
Review Comment:
Hey @1996fanrui , I see the following options w.r.t ChangelogStateBackend:
1. Add a new version of `StateMetaInfoReader` and explicitly encode whether the state is null-tolerant
2. Add a new null-tolerant operation along with `ADD_OR_UPDATE_ELEMENT`
3. Ignore the problem (leave Changelog State Backend inconsistent with other backends) and maybe keep a ticket to unify it
I didn't think through the 1st option though. One disadvantage (or advantage?) is that it'd be relevant to all state backends.
I personally don't have any preference.
--
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] [FLINK-38137][state] RocksDB and ForSt state backend are compatible with the case where the user value of the map state is null [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #26831:
URL: https://github.com/apache/flink/pull/26831#issuecomment-3117442949
@flinkbot run azure
--
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] [FLINK-38137][state] RocksDB and ForSt state backend are compatible with the case where the user value of the map state is null [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #26831:
URL: https://github.com/apache/flink/pull/26831#discussion_r2230489782
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java:
##########
@@ -176,7 +176,10 @@ <T> byte[] serializeValueNullSensitive(T value, TypeSerializer<T> serializer)
throws IOException {
dataOutputView.clear();
dataOutputView.writeBoolean(value == null);
- return serializeValueInternal(value, serializer);
+ if (value != null) {
+ serializer.serialize(value, dataOutputView);
+ }
+ return dataOutputView.getCopyOfBuffer();
Review Comment:
Hey @rkhachatryan , I found the change log state backend is serialized in https://issues.apache.org/jira/browse/FLINK-21355 , hope to hear your opinion as well.
Thank you both in advance.
--
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] [FLINK-38137][state] RocksDB and ForSt state backend are compatible with the case where the user value of the map state is null [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui merged PR #26831:
URL: https://github.com/apache/flink/pull/26831
--
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