Jackeyzhe commented on code in PR #29218:
URL: https://github.com/apache/flink/pull/29218#discussion_r4095353734
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java:
##########
@@ -188,24 +188,29 @@ boolean runOnce() {
}
// execute the task outside of lock, so that it can be woken up
- boolean taskFinished;
+ boolean taskFinished = false;
+ boolean taskRunCompleted = false;
try {
taskFinished = task.run();
+ taskRunCompleted = true;
} catch (Exception e) {
throw new RuntimeException(
String.format(
"SplitFetcher thread %d received unexpected
exception while polling the records",
id),
e);
- }
-
- // re-acquire lock as all post-processing steps, need it
- lock.lock();
- try {
- this.runningTask = null;
- processTaskResultUnsafe(task, taskFinished);
} finally {
- lock.unlock();
+ // Re-acquire the lock because clearing the current task and
processing a successful
+ // result must be atomic with respect to wakeup and shutdown.
+ lock.lock();
+ try {
+ this.runningTask = null;
Review Comment:
Clearing runningTask after task.run() fails can make isIdle() true before
the exceptional run() path finishes. The manager may then remove the fetcher
via shutdown(true), which does not release recordsProcessedLatch; no
synchronization batch is enqueued on failure, and a later manager close cannot
reach the removed fetcher. Please cover this failure/reap interleaving and
ensure the latch can be released.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
