You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "minxhe (via GitHub)" <gi...@apache.org> on 2026/10/22 22:37:11 UTC
[PR] [FLINK-41094][state/forst] Fail async map access on iterator errors and close iterators by owner only [flink]
minxhe opened a new pull request, #29574:
URL: https://github.com/apache/flink/pull/29574
## What is the purpose of the change
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`, async `MapState` operations had two related iterator bugs:
1. **Silent read-error truncation:** `ForStDBIterRequest` (map iteration), `ForStDBMapCheckRequest` (`asyncIsEmpty`), and `ForStDBBunchPutRequest` (`clear`) treated every `!isValid()` as normal exhaustion without checking `status()`, so a read error silently produced a truncated page, a false "empty" result, or a partial clear.
2. **Leaked or cross-thread-closed iterators on failure:**
- `ForStIterateOperation` completed failed or skipped requests exceptionally without closing their `ForStDBIterRequest`, leaking native `RocksIterator` handles.
- `ForStStateExecutor#executeBatchRequests` closed every `ForStDBIterRequest` in `iterRequests` on whichever thread failed the batch, racing sibling worker threads still reading from their iterators or closing iterators already handed off to completed pages.
This PR checks `status()` whenever an iterator becomes invalid, closes failed/skipped iterate requests on their owning worker thread in `ForStIterateOperation`, and removes the cross-thread close loop from `ForStStateExecutor`.
Jira: https://issues.apache.org/jira/browse/FLINK-41094
## Brief change log
- Check `rocksIterator.status()` when `!rocksIterator.isValid()` in `ForStDBIterRequest`, `ForStDBMapCheckRequest`, and `ForStDBBunchPutRequest`, failing the request on `RocksDBException`.
- Close `ForStDBIterRequest` on its owning worker thread in `ForStIterateOperation#failRequest` (attaching any close exception as suppressed) on both the catch path and the skipped-continuation path.
- Remove the cross-thread `iterRequest.close()` loop in `ForStStateExecutor#executeBatchRequests`.
- Add regression tests in `ForStDBIterateOperationTest` and `ForStStateExecutorTest`.
## Verifying this change
This change added tests and can be verified as follows:
- On unmodified `master` (`8d01380ff5bf9892462ef9808b1b56fd9723f54d`), all 8 new test invocations across `ForStDBIterateOperationTest` and `ForStStateExecutorTest` fail (leaked native iterator handles on worker failure/skip, silent success on `RocksIterator#status()` errors, and cross-thread iterator closure).
- With the fix, all 15 tests across `ForStDBIterateOperationTest` and `ForStStateExecutorTest` pass.
```sh
./mvnw -B -ntp -pl flink-state-backends/flink-statebackend-forst -Dflink.forkCountUnitTest=1 \
-Dtest=ForStDBIterateOperationTest,ForStStateExecutorTest test
# 15 run, 0 failures
./mvnw -B -ntp -pl flink-state-backends/flink-statebackend-forst \
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 (async ForSt `MapState` iterator status check and failure cleanup)
- 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41094][state/forst] Fail async map access on iterator errors and close iterators by owner only [flink]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #29574:
URL: https://github.com/apache/flink/pull/29574#discussion_r4354495105
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java:
##########
@@ -305,6 +311,14 @@ public void shutdown() {
if (!sharedWriteThread) {
shutdownAndWait(writeThreads);
}
+ ForStDBIterRequest<?, ?, ?, ?, ?> iterRequest;
+ while ((iterRequest = cleanupIterRequests.poll()) != null) {
+ try {
+ iterRequest.close();
Review Comment:
This cleanup can still race an active reader when shutdown is interrupted. `shutdownAndWait()` catches `InterruptedException`, calls `shutdownNow()`, and returns without confirming termination. A worker inside native iterator I/O need not stop on a Java interrupt, so this loop can free an iterator that the worker is still using. Continue waiting for actual executor termination before draining the queue, and restore the shutdown thread's interrupt status afterward. Add a regression test that interrupts shutdown while a reader remains active and verifies that its iterator is not closed until the reader stops.
--
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-41094][state/forst] Fail async map access on iterator errors and close iterators by owner only [flink]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #29574:
URL: https://github.com/apache/flink/pull/29574#discussion_r4354286286
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java:
##########
@@ -216,14 +215,7 @@ public CompletableFuture<Void> executeBatchRequests(
coordinatorThread)
.exceptionally(
e -> {
- try {
- for (ForStDBIterRequest<?, ?, ?, ?, ?> iterRequest :
- iterRequests) {
- iterRequest.close();
- }
- } catch (IOException ioException) {
- LOG.error("Close iterRequests fail", ioException);
- }
+ // Iterators are closed only by their owner.
Review Comment:
Removing the batch cleanup leaves successful partial-page iterators without a cleanup path when a sibling request fails. If the held request in `testBatchFailureLeavesIteratorsToTheirOwners` has more than `CACHE_SIZE_LIMIT` entries, it returns an open iterator after the batch fails. `executionError` then rejects further loading requests, and neither `ForStMapIterator` nor executor shutdown closes the retained handle. The worker-side `failRequest` does not cover this successful request. Add owner-safe reclamation for outstanding iterators, for example by tracking them and closing them during shutdown only after all workers have stopped. Keep the cross-thread close loop removed, and extend the regression test to cover a partial page and eventual handle cleanup.
--
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-41094][state/forst] Fail async map access on iterator errors and close iterators by owner only [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29574:
URL: https://github.com/apache/flink/pull/29574#issuecomment-6352937610
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3c61cfb69954016af2bba27ac0aef9393efa86b7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3c61cfb69954016af2bba27ac0aef9393efa86b7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3c61cfb69954016af2bba27ac0aef9393efa86b7 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