jerrypeng commented on PR #57092: URL: https://github.com/apache/spark/pull/57092#issuecomment-4986285381
@mridulm > Rest sound good to me, thanks @jerrypeng ! On output commit: > > > > In other words, cleanup is triggered by result stage completion, and tied to commit -> which is what happens today as well ? > > > I am trying to see what I am missing here - I dont know enough about streaming nuances to know if there are any sub cases I might be missing ! > > > > > > The distinction I'm drawing is that a PG is an atomic scheduling unit — there are no per-task or per-partition replays within it. That matters because OutputCommitCoordinator is built on the opposite assumption: it arbitrates commits per task attempt, authorizing exactly one attempt per partition and denying any later request for a partition that has already committed. > > That's sound today because a committed partition is never recomputed — a stage rerun (e.g. on fetch failure) only recomputes the missing partitions, so a task that already committed never needs to commit again, and the permanent "deny" is exactly right. A PG breaks that assumption. Because the group is atomic, a failure anywhere reruns the whole group — including a result stage whose tasks already succeeded and committed. So committed partitions do get rerun, and those rerun tasks must be allowed to commit again — but the coordinator still holds the previous attempt's authorized committers and would deny them (success clears nothing; only a failed holder's slot is cleared today). > > So to let the rerun's tasks commit, we either (a) rerun the members under fresh stage ids — a fresh coordinator StageState with no prior committers — or (b) reset the committed state for those stages in the OutputCommitCoordinator (e.g. stageEnd on teardown) before the rerun. > > There are nuances around which task 'wins' and/or is 'promoted' when there are failures/concurrent executions in output commiters. But - those aspects are, as already detailed in the spec, explicitly not supported for PG -> that is, we can only have a single task committing for a partition. In other words, there is no speculative execution in scope, and there is no task retries in case of failures - so those aspects of commit protocol wont apply for ResultStage within PG. If/when a task fails, the entire job is failed -> which results in failing the commit as well as it gets aborted. > > That is, I dont see a reason why existing committers would not work - it is just a stronger contract than what committers need to handle today. On other hand, if commit is denied (for whatever reason) - that is indeed a reason to fail the task -> and so the PG. > > In other words, based on what you are describing - all existing committers should continue to work ? That should significantly improve interop with existing ecosystem. Am I missing something here ?. Yup, you're right — existing committers work unchanged in v1. I should've been clearer about v1 vs. post-v1 (I've already updated the spec for this). In v1 this isn't an issue: any failure fails the whole job and the caller reruns it as a new job, so re-commits land on fresh stage ids with no prior committers. It only matters post-v1, if the scheduler reruns a PG in place instead of failing the job — which is the only case that reuses a stage id while the coordinator still holds the prior committers. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
