pnowojski commented on code in PR #22806:
URL: https://github.com/apache/flink/pull/22806#discussion_r1251679541
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java:
##########
@@ -220,8 +223,20 @@ void sendEventToSourceOperator(int subtaskId,
OperatorEvent event) {
String.format("Failed to send event %s to subtask %d", event,
subtaskId));
}
- ScheduledExecutorService getCoordinatorExecutor() {
- return coordinatorExecutor;
+ void sendEventToSourceOperatorIfTaskReady(int subtaskId, OperatorEvent
event) {
+ checkSubtaskIndex(subtaskId);
+
+ callInCoordinatorThread(
+ () -> {
+ final OperatorCoordinator.SubtaskGateway gateway =
+
subtaskGateways.getOnlyGatewayAndNotCheckReady(subtaskId);
+ if (gateway != null) {
+ gateway.sendEvent(event);
+ }
Review Comment:
> But the corresponding subtask may be restarting, causing
gateway.sendEvent(event) to fail, right?
Yes, that's my concern.
Re your 2nd and 3rd paragraphs:
If `sendEvent` fails, it will throw an exception, that will bubble up until
the catch-all safety net in `schedulePeriodTask`, so that will cause the job to
failover. Which I guess is currently mostly fine, unless that source has been
used in batch job with multiple failover regions. In that case expected
behaviour should be that only subtasks belonging to the failover region with
source subtasks should restart, not whole job. That's my main worry with this
race condition. That's why I suggested:
> Shouldn't we have a try/catch here and actually checking if the exception
is "retry-able", so things like "subtask has failed" or "subtask not ready" or
"task is initialising"? 🤔
--
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]
