SAlexandru commented on code in PR #29218:
URL: https://github.com/apache/flink/pull/29218#discussion_r4105525736
##########
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:
thanks ! should be fixed now by adding a failed flag and added a test
covering this
--
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]
