Aaron He created FLINK-41094:
--------------------------------
Summary: ForSt async MapState operations ignore
RocksIterator#status() on invalid iterators and close batch iterators across
threads
Key: FLINK-41094
URL: https://issues.apache.org/jira/browse/FLINK-41094
Project: Flink
Issue Type: Bug
Components: Runtime / State Backends
Reporter: Aaron He
h2. Problem
1. In RocksDB/ForSt, an invalid RocksIterator (isValid() == false) indicates
either normal end-of-data or a read/I/O error; only RocksIterator#status()
distinguishes the two. In flink-statebackend-forst, three async MapState
request types treat every !isValid() as normal exhaustion without checking
status():
* ForStDBIterRequest#process sets encounterEnd = true and returns a truncated
or empty page as a normal terminal iterator.
* ForStDBMapCheckRequest#process completes asyncIsEmpty() as empty when seek
fails with an invalid iterator.
* ForStDBBunchPutRequest#process (map clear) exits its prefix-scan loop when
isValid() becomes false and commits a partial delete batch.
2. When an execution batch fails in ForStStateExecutor#executeBatchRequests,
the exceptionally() handler loops over every ForStDBIterRequest in the batch
and calls iterRequest.close() on the failing thread, racing sibling read-thread
workers that are still accessing their iterators or closing iterators already
handed off to completed pages.
h2. Proposal
* Check rocksIterator.status() whenever isValid() is false at the end of
iteration in ForStDBIterRequest, ForStDBMapCheckRequest, and
ForStDBBunchPutRequest, failing the request if status() throws RocksDBException.
* Close each ForStDBIterRequest only on its owning worker thread in
ForStIterateOperation when a request fails or is skipped after an earlier
error, and remove the cross-thread iterRequests close loop in
ForStStateExecutor#executeBatchRequests.
h2. Public source
[ForStDBIterRequest.java|https://github.com/apache/flink/blob/855597a5460630eb8f72510277bfed7c0be0a004/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBIterRequest.java#L129-L148]
[ForStDBMapCheckRequest.java|https://github.com/apache/flink/blob/855597a5460630eb8f72510277bfed7c0be0a004/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBMapCheckRequest.java#L60-L71]
[ForStDBBunchPutRequest.java|https://github.com/apache/flink/blob/855597a5460630eb8f72510277bfed7c0be0a004/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java#L62-L71]
[ForStStateExecutor.java|https://github.com/apache/flink/blob/855597a5460630eb8f72510277bfed7c0be0a004/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java#L217-L227]
h2. Tests and acceptance
{noformat}
./mvnw -pl flink-state-backends/flink-statebackend-forst
-Dtest=ForStStateExecutorTest,ForStDBIterateOperationTest test
{noformat}
h2. Proposed release note
ForSt async MapState iteration, isEmpty, and clear now propagate underlying
RocksIterator status errors instead of returning truncated results, and
iterators are closed only by their owning worker.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
