[
https://issues.apache.org/jira/browse/FLINK-39624?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18080201#comment-18080201
]
Zhen Zhang commented on FLINK-39624:
------------------------------------
Hi [~cbakos]
Thanks for the detailed report. Would you like to submit a PR for this issue?
> `TtlListState.asyncGet()` returns a non-null wrapper around a null inner
> iterator, causing NPE in `onNext`/`isEmpty` when the state is empty
> --------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-39624
> URL: https://issues.apache.org/jira/browse/FLINK-39624
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Async State Processing, Runtime / State
> Backends
> Affects Versions: 2.0.0, 2.0-preview, 2.1.0, 2.0.1, 2.2.0, 2.1.1, 2.3.0,
> 2.4.0
> Environment: - Flink 2.2.0
> - ForSt state backend
> - Async State V2 enabled
> - JDK 17
> Reporter: Csanád Bakos
> Priority: Major
>
> `AppendingState#asyncGet()` documents the contract:
> bq. NOTE TO IMPLEMENTERS: if the state is empty, then this method should
> return null wrapped by a StateFuture.
> The ForSt backend honors this contract.
> `ForStDBListGetRequest.completeStateFuture` (release-2.2.0):
> {code:java}
> public void completeStateFuture(byte[] bytesValue) throws IOException {
> if (bytesValue == null) {
> future.complete(null); // null per the AppendingState contract
> return;
> }
> ...
> {code}
> `TtlListState.asyncGet()` does *not* propagate the null — it wraps
> unconditionally:
> {code:java}
> @Override
> public StateFuture<StateIterator<T>> asyncGet() {
> // 1. The timestamp of elements in list state isn't updated when get even
> if updateTsOnRead is true.
> // 2. we don't clear state here cause forst is LSM-tree based.
> return original.asyncGet().thenApply(stateIter -> new
> AsyncIteratorWrapper(stateIter));
> }
> {code}
> `AsyncFutureImpl#thenApply` invokes the lambda even when the input value is
> `null` (no null short-circuit), so when the underlying state is empty,
> callers receive a *non-null* `AsyncIteratorWrapper` whose private
> `originalIterator` field is `null`.
> Both methods of the wrapper then NPE when invoked:
> {code:java}
> public StateFuture<Void> onNext(ThrowingConsumer<T, Exception> iterating) {
> ThrowingConsumer<TtlValue<T>, Exception> ttlIterating = (item) -> { ... };
> return originalIterator.onNext(ttlIterating); // NPE on line 221
> }
> public boolean isEmpty() {
> return originalIterator.isEmpty(); // NPE on line 226
> }
> {code}
> Notice that the *synchronous* counterpart `TtlListState#get()` correctly
> null-checks:
> {code:java}
> public Iterable<T> get() {
> Iterable<TtlValue<T>> ttlValue = original.get();
> ttlValue = ttlValue == null ? Collections.emptyList() : ttlValue;
> ...
> {code}
> So the bug is purely in the async path of `TtlListState`.
> h2. Steps to Reproduce
> * A `KeyedProcessFunction` (or any keyed operator) with async state V2
> enabled, ForSt backend, on a `ListState<E>` whose descriptor has
> `enableTimeToLive(ttlConfig)` set: Without ever calling `asyncAdd`, call
> `state.asyncGet().thenCompose(iter -> iter.onNext(consumer))`.
> * Or, call `asyncGet` after a previous `asyncClear()` (ForSt's `Delete(key)`
> is `Put(key, null)`, so the next read is `null`).
> Either path throws:
> {noformat}
> java.lang.NullPointerException
> at
> org.apache.flink.runtime.state.v2.ttl.TtlListState$AsyncIteratorWrapper.onNext(TtlListState.java:221)
> at
> org.apache.flink.core.asyncprocessing.AsyncFutureImpl.lambda$thenCompose$6(AsyncFutureImpl.java:163)
> at
> org.apache.flink.runtime.asyncprocessing.AsyncFutureFactory.lambda$create$0(AsyncFutureFactory.java:49)
> at
> org.apache.flink.runtime.asyncprocessing.CallbackRunnerWrapper.lambda$submit$0(CallbackRunnerWrapper.java:57)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
> at
> org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:118)
> at
> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:415)
> ...
> {noformat}
> h2. Expected Behavior
> Per the `AppendingState#asyncGet` javadoc, `iter` is `null` when state is
> empty, so `if (iter == null)` (or `StateFutureUtils.toIterable`'s null guard)
> is sufficient to handle the empty case.
> h2. Actual Behavior
> `iter` is a non-null `AsyncIteratorWrapper` wrapping a null inner iterator.
> Any subsequent call NPEs.
> h2. Suggested Fix
> Propagate the null in `TtlListState.asyncGet`:
> {code:java}
> @Override
> public StateFuture<StateIterator<T>> asyncGet() {
> return original.asyncGet().thenApply(stateIter ->
> stateIter == null ? null : new AsyncIteratorWrapper(stateIter));
> }
> {code}
> This mirrors the synchronous `get()` and brings the async path back in line
> with the `AppendingState` contract.
> h2. Source References (release-2.2.0)
> - Buggy method:
> https://github.com/apache/flink/blob/release-2.2.0/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/ttl/TtlListState.java#L67-L72
> - NPE-prone wrapper:
> https://github.com/apache/flink/blob/release-2.2.0/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/ttl/TtlListState.java#L189-L228
> - Sync `get()` for comparison:
> https://github.com/apache/flink/blob/release-2.2.0/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/ttl/TtlListState.java#L79-L85
> - Contract source (ForSt):
> https://github.com/apache/flink/blob/release-2.2.0/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBListGetRequest.java#L46-L50
> - Contract spec:
> https://github.com/apache/flink/blob/release-2.2.0/flink-core-api/src/main/java/org/apache/flink/api/common/state/v2/AppendingState.java#L40-L51
> - `thenApply` does not short-circuit on null:
> https://github.com/apache/flink/blob/release-2.2.0/flink-core/src/main/java/org/apache/flink/core/asyncprocessing/AsyncFutureImpl.java#L60-L92
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
