Repository: spark
Updated Branches:
refs/heads/master ebf4bfb96 -> 290c30a53
[SPARK-24470][CORE] RestSubmissionClient to be robust against 404 & non json
responses
## What changes were proposed in this pull request?
Added check for 404, to avoid json parsing on not found response and to avoid
returning malformed or bad request when it was a not found http response.
Not sure if I need to add an additional check on non json response
[if(connection.getHeaderField("Content-Type").contains("text/html")) then
exception] as non-json is a subset of malformed json and covered in flow.
## How was this patch tested?
./dev/run-tests
Author: Rekha Joshi <[email protected]>
Closes #21684 from rekhajoshm/SPARK-24470.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/290c30a5
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/290c30a5
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/290c30a5
Branch: refs/heads/master
Commit: 290c30a53fc2f46001846ab8abafcc69b853ba98
Parents: ebf4bfb
Author: Rekha Joshi <[email protected]>
Authored: Wed Jul 11 13:48:28 2018 -0500
Committer: Sean Owen <[email protected]>
Committed: Wed Jul 11 13:48:28 2018 -0500
----------------------------------------------------------------------
.../deploy/rest/RestSubmissionClient.scala | 60 ++++++++++++--------
1 file changed, 37 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/290c30a5/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
----------------------------------------------------------------------
diff --git
a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
index 742a958..31a8e3e 100644
---
a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
+++
b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
@@ -233,30 +233,44 @@ private[spark] class RestSubmissionClient(master: String)
extends Logging {
private[rest] def readResponse(connection: HttpURLConnection):
SubmitRestProtocolResponse = {
import scala.concurrent.ExecutionContext.Implicits.global
val responseFuture = Future {
- val dataStream =
- if (connection.getResponseCode == HttpServletResponse.SC_OK) {
- connection.getInputStream
- } else {
- connection.getErrorStream
+ val responseCode = connection.getResponseCode
+
+ if (responseCode != HttpServletResponse.SC_OK) {
+ val errString =
Some(Source.fromInputStream(connection.getErrorStream())
+ .getLines().mkString("\n"))
+ if (responseCode == HttpServletResponse.SC_INTERNAL_SERVER_ERROR &&
+ !connection.getContentType().contains("application/json")) {
+ throw new SubmitRestProtocolException(s"Server responded with
exception:\n${errString}")
+ }
+ logError(s"Server responded with error:\n${errString}")
+ val error = new ErrorResponse
+ if (responseCode == RestSubmissionServer.SC_UNKNOWN_PROTOCOL_VERSION) {
+ error.highestProtocolVersion = RestSubmissionServer.PROTOCOL_VERSION
+ }
+ error.message = errString.get
+ error
+ } else {
+ val dataStream = connection.getInputStream
+
+ // If the server threw an exception while writing a response, it will
not have a body
+ if (dataStream == null) {
+ throw new SubmitRestProtocolException("Server returned empty body")
+ }
+ val responseJson = Source.fromInputStream(dataStream).mkString
+ logDebug(s"Response from the server:\n$responseJson")
+ val response = SubmitRestProtocolMessage.fromJson(responseJson)
+ response.validate()
+ response match {
+ // If the response is an error, log the message
+ case error: ErrorResponse =>
+ logError(s"Server responded with error:\n${error.message}")
+ error
+ // Otherwise, simply return the response
+ case response: SubmitRestProtocolResponse => response
+ case unexpected =>
+ throw new SubmitRestProtocolException(
+ s"Message received from server was not a
response:\n${unexpected.toJson}")
}
- // If the server threw an exception while writing a response, it will
not have a body
- if (dataStream == null) {
- throw new SubmitRestProtocolException("Server returned empty body")
- }
- val responseJson = Source.fromInputStream(dataStream).mkString
- logDebug(s"Response from the server:\n$responseJson")
- val response = SubmitRestProtocolMessage.fromJson(responseJson)
- response.validate()
- response match {
- // If the response is an error, log the message
- case error: ErrorResponse =>
- logError(s"Server responded with error:\n${error.message}")
- error
- // Otherwise, simply return the response
- case response: SubmitRestProtocolResponse => response
- case unexpected =>
- throw new SubmitRestProtocolException(
- s"Message received from server was not a
response:\n${unexpected.toJson}")
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
