[SPARK-24063][SS] Control maximum epoch backlog for ContinuousExecution #21392
Conversation
|
@jose-torres Hi Jose, could you take a look at this pr please? I had doubts how to properly implement error reporting logic we discussed and this is what I came up with. |
| val maxBacklogExceeded = epochEndpoint.askSync[Boolean](CheckIfMaxBacklogIsExceeded) | ||
| if (maxBacklogExceeded) { | ||
| throw new IllegalStateException( | ||
| "Size of the epochs queue has exceeded maximum allowed epoch backlog.") |
There was a problem hiding this comment.
Throw exception will make epochUpdateThread stop working, but the application will keep working?
I think it's better to block and wait old epoch to be committed.
There was a problem hiding this comment.
Agreed that the code as written won't shut down the stream. But I think it does make sense to kill the stream rather than waiting for old epochs. If we end up with a large backlog it's almost surely because some partition isn't making any progress, so I wouldn't expect Spark to ever be able to catch up.
| } else { | ||
| logDebug(s"Epoch $epoch has received commits from all partitions " + | ||
| s"and is waiting for epoch ${epoch - 1} to be committed first.") | ||
| epochsWaitingToBeCommitted.add(epoch) |
There was a problem hiding this comment.
once maxEpochBacklogExceeded is set to true, can never be set to false again?
There was a problem hiding this comment.
Basing on what I discussed with Jose the stream should be killed if backlog exceeds value of a certain config option, so yes, why set it back to false later. At least that's how I see it
There was a problem hiding this comment.
The simplest way to test the changes would be to use a TestSparkSession with too few cores - e.g. local(1) but 2 input partitions. Then you can start a stream with something like a 1 ms checkpoint interval and max backlog 10 to quickly get a failure. (All epochs will end up in the backlog if there aren't enough cores to schedule one of the partitions, since no global commit will ever succeed.)
| /** | ||
| * Returns boolean indicating if size of the epochs queue has exceeded maximum epoch backlog. | ||
| */ | ||
| private[sql] case object CheckIfMaxBacklogIsExceeded extends EpochCoordinatorMessage |
There was a problem hiding this comment.
I'm not sure we need to make a side-channel in the RPC handler for this. I'd try to just make the query fail when the condition is reached in the first place.
There was a problem hiding this comment.
Do you mean make the query fail right from EpochCoordinator? If yes, I wanted to do so, but didn't figure out how to terminate query with exception.
EpochCoordinator has query: ContinuousExecution as a parameter, but then I don't see a suitable method for query.. Closest I found is stop() I guess.
Or am I looking in a completely wrong direction? Please give a hint.
There was a problem hiding this comment.
I think we'd probably want to add some method like private[streaming] stopWithException(e) to ContinuousExecution.
There was a problem hiding this comment.
Okay, thought about something like this but wasn't sure if it's fine to do so in scope of this change. Thanks
|
Hi all, any update on this PR? |
|
@HyukjinKwon Hi, I've stopped working on it for some time now |
|
In that case, would you mind if I ask to leave this closed for now and reopen when you start to work on this? I am trying to leave active PRs only. |
|
Sure, no problem. |
|
Can one of the admins verify this patch? |
What changes were proposed in this pull request?
This pull request adds maxEpochBacklog SQL configuration option. EpochCoordinator tracks if the length of the queue of waiting epochs has exceeded it. Upon this condition stream is stopped with an error indicating too many epochs stacked up
How was this patch tested?
Existing unit tests