Skip to content

Commit addcecb

Browse files
codeboyyongcodeboyyong
committed
[SPARK-1516]Throw exception in yarn client instead of run system.exit directly.
 All the changes is in the package of "org.apache.spark.deploy.yarn": 1) Throw IllegalArgumentException in ClinetArguments instead of exit directly. 2) In Client's main method, if exception is caught, it will exit with code 1, otherwise exit with code 0. 3) In YarnClientSchedulerBackend's start method, if IllegalArgumentException is caught, it will exit with code 1, otherwise throw that exception. 4) Fix some message typo in the Client.scala After the fix, if user integrate the spark yarn client into their applications, when the argument is wrong or the running is finished, the application won't be terminated.
1 parent 706e38f commit addcecb

File tree

4 files changed

+63
-36
lines changed

4 files changed

+63
-36
lines changed

‎yarn/alpha/src/main/scala/org/apache/spark/deploy/yarn/Client.scala

Lines changed: 22 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,6 @@ class Client(args: ClientArguments, conf: Configuration, sparkConf: SparkConf)
9393
def run() {
9494
val appId = runApp()
9595
monitorApplication(appId)
96-
System.exit(0)
9796
}
9897

9998
def validateArgs() = {
@@ -109,7 +108,7 @@ class Client(args: ClientArguments, conf: Configuration, sparkConf: SparkConf)
109108
).foreach { case(cond, errStr) =>
110109
if (cond) {
111110
logError(errStr)
112-
args.printUsageAndExit(1)
111+
throw new IllegalArgumentException(args.getUsageMessage())
113112
}
114113
}
115114
}
@@ -135,17 +134,19 @@ class Client(args: ClientArguments, conf: Configuration, sparkConf: SparkConf)
135134

136135
def verifyClusterResources(app: GetNewApplicationResponse) = {
137136
val maxMem = app.getMaximumResourceCapability().getMemory()
138-
logInfo("Max mem capabililty of a single resource in this cluster " + maxMem)
137+
logInfo("Max mem capability of a single resource in this cluster " + maxMem)
139138

140139
// If we have requested more then the clusters max for a single resource then exit.
141140
if (args.workerMemory > maxMem) {
142-
logError("the worker size is to large to run on this cluster " + args.workerMemory)
143-
System.exit(1)
141+
val errorMessage = s"the worker size is too large to run on this cluster ${args.workerMemory}"
142+
logError(errorMessage)
143+
throw new IllegalArgumentException(errorMessage)
144144
}
145145
val amMem = args.amMemory + YarnAllocationHandler.MEMORY_OVERHEAD
146146
if (amMem > maxMem) {
147-
logError("AM size is to large to run on this cluster " + amMem)
148-
System.exit(1)
147+
val errorMessage = s"AM size is too large to run on this cluster $amMem"
148+
logError(errorMessage)
149+
throw new IllegalArgumentException(errorMessage)
149150
}
150151

151152
// We could add checks to make sure the entire cluster has enough resources but that involves
@@ -229,8 +230,9 @@ class Client(args: ClientArguments, conf: Configuration, sparkConf: SparkConf)
229230
val delegTokenRenewer = Master.getMasterPrincipal(conf)
230231
if (UserGroupInformation.isSecurityEnabled()) {
231232
if (delegTokenRenewer == null || delegTokenRenewer.length() == 0) {
232-
logError("Can't get Master Kerberos principal for use as renewer")
233-
System.exit(1)
233+
val errorMessage = "Can't get Master Kerberos principal for use as renewer"
234+
logError(errorMessage)
235+
throw new IllegalArgumentException(errorMessage)
234236
}
235237
}
236238
val dst = new Path(fs.getHomeDirectory(), appStagingDir)
@@ -475,9 +477,18 @@ object Client {
475477
System.setProperty("SPARK_YARN_MODE", "true")
476478

477479
val sparkConf = new SparkConf
478-
val args = new ClientArguments(argStrings, sparkConf)
479480

480-
new Client(args, sparkConf).run
481+
try {
482+
val args = new ClientArguments(argStrings, sparkConf)
483+
new Client(args, sparkConf).run()
484+
} catch {
485+
case e: Exception => {
486+
Console.err.println(e.getMessage)
487+
System.exit(1)
488+
}
489+
}
490+
491+
System.exit(0)
481492
}
482493

483494
// Based on code from org.apache.hadoop.mapreduce.v2.util.MRApps

‎yarn/common/src/main/scala/org/apache/spark/deploy/yarn/ClientArguments.scala

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -109,11 +109,11 @@ class ClientArguments(val args: Array[String], val sparkConf: SparkConf) {
109109

110110
case Nil =>
111111
if (userJar == null || userClass == null) {
112-
printUsageAndExit(1)
112+
throw new IllegalArgumentException(getUsageMessage())
113113
}
114114

115115
case _ =>
116-
printUsageAndExit(1, args)
116+
throw new IllegalArgumentException(getUsageMessage(args))
117117
}
118118
}
119119

@@ -122,11 +122,10 @@ class ClientArguments(val args: Array[String], val sparkConf: SparkConf) {
122122
}
123123

124124

125-
def printUsageAndExit(exitCode: Int, unknownParam: Any = null) {
126-
if (unknownParam != null) {
127-
System.err.println("Unknown/unsupported param " + unknownParam)
128-
}
129-
System.err.println(
125+
def getUsageMessage(unknownParam: Any = null): String = {
126+
val message = if (unknownParam != null) s"Unknown/unsupported param $unknownParam \n" else ""
127+
128+
message +
130129
"Usage: org.apache.spark.deploy.yarn.Client [options] \n" +
131130
"Options:\n" +
132131
" --jar JAR_PATH Path to your application's JAR file (required)\n" +
@@ -143,8 +142,7 @@ class ClientArguments(val args: Array[String], val sparkConf: SparkConf) {
143142
" --addJars jars Comma separated list of local jars that want SparkContext.addJar to work with.\n" +
144143
" --files files Comma separated list of files to be distributed with the job.\n" +
145144
" --archives archives Comma separated list of archives to be distributed with the job."
146-
)
147-
System.exit(exitCode)
145+
148146
}
149147

150148
}

‎yarn/common/src/main/scala/org/apache/spark/scheduler/cluster/YarnClientSchedulerBackend.scala

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -73,10 +73,17 @@ private[spark] class YarnClientSchedulerBackend(
7373
.foreach { case (optName, optParam) => addArg(optName, optParam, argsArrayBuf) }
7474

7575
logDebug("ClientArguments called with: " + argsArrayBuf)
76-
val args = new ClientArguments(argsArrayBuf.toArray, conf)
77-
client = new Client(args, conf)
78-
appId = client.runApp()
79-
waitForApp()
76+
try {
77+
val args = new ClientArguments(argsArrayBuf.toArray, conf)
78+
client = new Client(args, conf)
79+
appId = client.runApp()
80+
waitForApp()
81+
} catch {
82+
case e: IllegalArgumentException => {
83+
Console.err.println(e.getMessage)
84+
System.exit(1)
85+
}
86+
}
8087
}
8188

8289
def waitForApp() {

‎yarn/stable/src/main/scala/org/apache/spark/deploy/yarn/Client.scala

Lines changed: 23 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -113,7 +113,6 @@ class Client(args: ClientArguments, conf: Configuration, sparkConf: SparkConf)
113113
def run() {
114114
val appId = runApp()
115115
monitorApplication(appId)
116-
System.exit(0)
117116
}
118117

119118
// TODO(harvey): This could just go in ClientArguments.
@@ -130,7 +129,7 @@ class Client(args: ClientArguments, conf: Configuration, sparkConf: SparkConf)
130129
).foreach { case(cond, errStr) =>
131130
if (cond) {
132131
logError(errStr)
133-
args.printUsageAndExit(1)
132+
throw new IllegalArgumentException(args.getUsageMessage())
134133
}
135134
}
136135
}
@@ -160,15 +159,18 @@ class Client(args: ClientArguments, conf: Configuration, sparkConf: SparkConf)
160159

161160
// If we have requested more then the clusters max for a single resource then exit.
162161
if (args.workerMemory > maxMem) {
163-
logError("Required worker memory (%d MB), is above the max threshold (%d MB) of this cluster.".
164-
format(args.workerMemory, maxMem))
165-
System.exit(1)
162+
val errorMessage =
163+
"Required worker memory (%d MB), is above the max threshold (%d MB) of this cluster."
164+
.format(args.workerMemory, maxMem)
165+
logError(errorMessage)
166+
throw new IllegalArgumentException(errorMessage)
166167
}
167168
val amMem = args.amMemory + YarnAllocationHandler.MEMORY_OVERHEAD
168169
if (amMem > maxMem) {
169-
logError("Required AM memory (%d) is above the max threshold (%d) of this cluster".
170-
format(args.amMemory, maxMem))
171-
System.exit(1)
170+
val errorMessage = "Required AM memory (%d) is above the max threshold (%d) of this cluster"
171+
.format(args.amMemory, maxMem)
172+
logError(errorMessage)
173+
throw new IllegalArgumentException(errorMessage)
172174
}
173175

174176
// We could add checks to make sure the entire cluster has enough resources but that involves
@@ -244,8 +246,9 @@ class Client(args: ClientArguments, conf: Configuration, sparkConf: SparkConf)
244246
val delegTokenRenewer = Master.getMasterPrincipal(conf)
245247
if (UserGroupInformation.isSecurityEnabled()) {
246248
if (delegTokenRenewer == null || delegTokenRenewer.length() == 0) {
247-
logError("Can't get Master Kerberos principal for use as renewer")
248-
System.exit(1)
249+
val errorMessage = "Can't get Master Kerberos principal for use as renewer"
250+
logError(errorMessage)
251+
throw new IllegalArgumentException(errorMessage)
249252
}
250253
}
251254
val dst = new Path(fs.getHomeDirectory(), appStagingDir)
@@ -489,9 +492,17 @@ object Client {
489492
// see Client#setupLaunchEnv().
490493
System.setProperty("SPARK_YARN_MODE", "true")
491494
val sparkConf = new SparkConf()
492-
val args = new ClientArguments(argStrings, sparkConf)
495+
try {
496+
val args = new ClientArguments(argStrings, sparkConf)
497+
new Client(args, sparkConf).run()
498+
} catch {
499+
case e: Exception => {
500+
Console.err.println(e.getMessage)
501+
System.exit(1)
502+
}
503+
}
493504

494-
new Client(args, sparkConf).run()
505+
System.exit(0)
495506
}
496507

497508
// Based on code from org.apache.hadoop.mapreduce.v2.util.MRApps

0 commit comments

Comments
 (0)