[jira] [Created] (FLINK-41094) ForSt async MapState operations ignore RocksIterator#status() on invalid iterators and close batch iterators across threads

["Aaron He (Jira)" <ji...@apache.org>]

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)