Skip to content

SPARK-23660: Fix exception in yarn cluster mode when application ended fast #20807

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
gaborgsomogyi wants to merge 5 commits into from
Closed

SPARK-23660: Fix exception in yarn cluster mode when application ended fast #20807

Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
114ac05
SPARK-23660: Fix exception in yarn cluster mode when application ende…
gaborgsomogyi Mar 13, 2018
442cfb4
Pause user class thread until proper initialisation
gaborgsomogyi Mar 13, 2018
5b304d1
Used sparkContextPromise as lock
gaborgsomogyi Mar 16, 2018
284ed68
Race fix + comment move
gaborgsomogyi Mar 16, 2018
33ca59d
Remove unnecessary comment
gaborgsomogyi Mar 16, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -417,8 +417,11 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
}
}

private def sparkContextInitialized(sc: SparkContext) = {
private def sparkContextInitialized(sc: SparkContext) = synchronized {

@vanzin vanzin Mar 15, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Some other code in this class uses synchronization on this, so I think it would be better to synchronize on sparkContextPromise in this case.

@gaborgsomogyi gaborgsomogyi Mar 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Used sparkContextPromise as lock.

// Notify runDriver function that SparkContext is available
sparkContextPromise.success(sc)
// Pause the user class thread in order to make proper initialisation in runDriver function
wait()
}

private def registerAM(
Expand Down Expand Up @@ -497,6 +500,8 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
// if the user app did not create a SparkContext.
throw new IllegalStateException("User did not initialize spark context!")
}
// After initialisation notify user class thread to continue

@vanzin vanzin Mar 15, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: rest of the code uses American spelling ("initialization"), so this should be consistent.

@gaborgsomogyi gaborgsomogyi Mar 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed and switched to US spell checker.

@vanzin vanzin Mar 16, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You should remove this comment and add one to the resumeDriver method.

@gaborgsomogyi gaborgsomogyi Mar 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Moved.

synchronized { notify() }

@vanzin vanzin Mar 15, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since you have to do this in two places, I'd create a method (e.g. resumeDriver) close to where sparkContextInitialized is declared, so that it's easier to find the context of why this is needed.

@gaborgsomogyi gaborgsomogyi Mar 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Moved into resumeDriver function right below sparkContextInitialized.

userClassThread.join()
} catch {
case e: SparkException if e.getCause().isInstanceOf[TimeoutException] =>
Expand All @@ -506,6 +511,8 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
finish(FinalApplicationStatus.FAILED,
ApplicationMaster.EXIT_SC_NOT_INITED,
"Timed out waiting for SparkContext.")
} finally {
synchronized { notify() }
}
}

Expand Down