[FLINK-35157][runtime] Sources with watermark alignment get stuck once some subtasks finish #24757
Conversation
…e some subtasks finish
|
hi, @1996fanrui would you mind reviewing this for me when you have a moment? Thank you very much! |
There was a problem hiding this comment.
Hi @elon-X , I'm wondering if "NoMoreSplitsEvent" would be appropriate to check if a subtask is finished.
The Split can be discovered dynamically. IIRC, kafka source subtask won't be FINISHED even if it doesn't have split for Flink Streaming Job. (This subtask may be assigned split after adding new kafka partition)
I prefer Gyula proposed solution: The solution could be to send out a max watermark event once the sources finish or to exclude them from the source coordinator.
We need to find a proper code place(source is definitely finished) to send MaxTimestamp.
I guess
may be a suitable place.| context.sendEventToSourceOperatorIfTaskReady( | ||
| subtaskId, new WatermarkAlignmentEvent(maxAllowedWatermark)); | ||
| // when subtask have been finished, do not send event. | ||
| if (!context.hasNoMoreSplits(subtaskId)) { |
There was a problem hiding this comment.
Does NoMoreSplits mean subtask has been finished?
I'm afraid they are not same. For example, some subtask doesn't have split when the parallelism of kafka source is greater than kafka partition.
I'm not sure do we need this change. IIUC, inside of context.sendEventToSourceOperatorIfTaskReady has a check : do send event when gateway is not null.
It means coordinator doesn't send event to subtask when subtask is finished, right?
Please correct me if anything is wrong, thanks
There was a problem hiding this comment.
Hi @1996fanrui ,
First of all, thank you for your detailed review and response. I would like to share my thoughts:
1.Regarding the SourceOperator, initially, I considered the issue of code positioning. I noticed that before switching to the DataInputStatus.END_OF_INPUT state, the SourceOperator receives a NoMoreSplitsEvent, and the SourceCoordinator also needs to determine whether the subtask has finished or not. Therefore, I chose to send the maximum timestamp within the if (event instanceof NoMoreSplitsEvent) { } I believe these two locations should be equivalent.
2.On the SourceCoordinator side, if a task has finished, the context.sendEventToSourceOperatorIfTaskReady code still sends events to that subtask. In the code provided by Gyula, if the response result from gateway.sendEvent(event).get() is obtained, the following exception occurs:
Caused by: java.util.concurrent.ExecutionException: org.apache.flink.runtime.operators.coordination.TaskNotRunningException: "Source: Sequence Source -> Filter -> Sink: Print to Std. Out (1/2)" is not running, but in state FINISHED
Therefore, I added the condition !context.hasNoMoreSplits(subtaskId). I also tested the scenario "some subtask doesn't have split when the parallelism of Kafka source is greater than Kafka partition," and indeed, no NoMoreSplitsEvent is sent.
Please correct me if my understanding is wrong. I will modify the code based on the final discussion results and submit it~
What is the purpose of the change
Sources with watermark alignment get stuck once some subtasks finish, this PR solves this problem.
Brief change log
while some subtasks have been finished, the SourceOperator send Long.MAX_VALUE to SourceCoordinator, and SourceCoordinator checks whether subtasks have been finished before sending the event.
Verifying this change
This change added tests and can be verified as follows:
Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation