Skip to content
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

[Spark-2381] stop the streaming application if the receiver job failed #1693

Closed
wants to merge 4 commits into from
Closed

[Spark-2381] stop the streaming application if the receiver job failed #1693

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f4660c5
[SPARK-1998] SparkFlumeEvent with body bigger than 1020 bytes are not…
joyyoj Jun 3, 2014
3f4a602
Merge remote-tracking branch 'remotes/apache/master'
joyyoj Jul 6, 2014
22e7821
Merge remote-tracking branch 'apache/master'
joyyoj Jul 31, 2014
0a156c5
SPARK-2381 Exit if the receiver job failed
joyyoj Jul 31, 2014
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
10 changes: 9 additions & 1 deletion streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -272,7 +272,15 @@ class ReceiverTracker(ssc: StreamingContext) extends Logging {

// Distribute the receivers and start them
logInfo("Starting " + receivers.length + " receivers")
ssc.sparkContext.runJob(tempRDD, startReceiver)
try {
ssc.sparkContext.runJob(tempRDD, startReceiver)
} catch {
case e : Throwable =>
Copy link
Contributor

Choose a reason for hiding this comment

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

Scala uses throwable for control flow, so throwable should never be caught in a Scala program. Please catch Exception

logError("receiver job crashed, detail:" + e.getStackTraceString)
Copy link
Contributor

Choose a reason for hiding this comment

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

Actually there is a better way to report this error. streamingContext.awaitTermination() can be made to throw the exception. Call waiter.notifyError(exception) which will unblock streamingContext.awaitTermination() with the exception and let the user decide whether to stop-and-exit or not.

// find a better way to stop the application ? make other stages depend on tempRdd ?
ssc.stop()
System.exit(1)
Copy link
Member

Choose a reason for hiding this comment

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

Please don't System.exit here

}
logInfo("All of the receivers have been terminated")
}

Expand Down