[SPARK-6980] [CORE] Akka timeout exceptions indicate which conf controls them (RPC Layer) #6205
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,8 +18,10 @@ | |
| package org.apache.spark.rpc | ||
|
|
||
| import java.net.URI | ||
| import java.util.concurrent.TimeoutException | ||
|
|
||
| import scala.concurrent.{Awaitable, Await, Future} | ||
| import scala.concurrent.duration._ | ||
| import scala.language.postfixOps | ||
|
|
||
| import org.apache.spark.{SecurityManager, SparkConf} | ||
|
|
@@ -66,7 +68,7 @@ private[spark] object RpcEnv { | |
| */ | ||
| private[spark] abstract class RpcEnv(conf: SparkConf) { | ||
|
|
||
| private[spark] val defaultLookupTimeout = RpcUtils.lookupRpcTimeout(conf) | ||
|
|
||
| /** | ||
| * Return RpcEndpointRef of the registered [[RpcEndpoint]]. Will be used to implement | ||
|
|
@@ -94,7 +96,7 @@ private[spark] abstract class RpcEnv(conf: SparkConf) { | |
| * Retrieve the [[RpcEndpointRef]] represented by `uri`. This is a blocking action. | ||
| */ | ||
| def setupEndpointRefByURI(uri: String): RpcEndpointRef = { | ||
| defaultLookupTimeout.awaitResult(asyncSetupEndpointRefByURI(uri)) | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -182,3 +184,106 @@ private[spark] object RpcAddress { | |
| RpcAddress(host, port) | ||
| } | ||
| } | ||
|
|
||
|
|
||
| /** | ||
| * An exception thrown if RpcTimeout modifies a [[TimeoutException]]. | ||
| */ | ||
| private[rpc] class RpcTimeoutException(message: String, cause: TimeoutException) | ||
| extends TimeoutException(message) { initCause(cause) } | ||
|
|
||
|
|
||
| /** | ||
| * Associates a timeout with a description so that a when a TimeoutException occurs, additional | ||
| * context about the timeout can be amended to the exception message. | ||
| * @param duration timeout duration in seconds | ||
| * @param timeoutProp the configuration property that controls this timeout | ||
| */ | ||
| private[spark] class RpcTimeout(val duration: FiniteDuration, val timeoutProp: String) { | ||
|
|
||
| /** Amends the standard message of TimeoutException to include the description */ | ||
| private def createRpcTimeoutException(te: TimeoutException): RpcTimeoutException = { | ||
| new RpcTimeoutException(te.getMessage() + ". This timeout is controlled by " + timeoutProp, te) | ||
| } | ||
|
|
||
| /** | ||
| * PartialFunction to match a TimeoutException and add the timeout description to the message | ||
| * | ||
| * @note This can be used in the recover callback of a Future to add to a TimeoutException | ||
| * Example: | ||
| * val timeout = new RpcTimeout(5 millis, "short timeout") | ||
| * Future(throw new TimeoutException).recover(timeout.addMessageIfTimeout) | ||
| */ | ||
| def addMessageIfTimeout[T]: PartialFunction[Throwable, T] = { | ||
| // The exception has already been converted to a RpcTimeoutException so just raise it | ||
| case rte: RpcTimeoutException => throw rte | ||
| // Any other TimeoutException get converted to a RpcTimeoutException with modified message | ||
| case te: TimeoutException => throw createRpcTimeoutException(te) | ||
| } | ||
|
|
||
| /** | ||
| * Wait for the completed result and return it. If the result is not available within this | ||
| * timeout, throw a [[RpcTimeoutException]] to indicate which configuration controls the timeout. | ||
| * @param awaitable the `Awaitable` to be awaited | ||
| * @throws RpcTimeoutException if after waiting for the specified time `awaitable` | ||
| * is still not ready | ||
| */ | ||
| def awaitResult[T](awaitable: Awaitable[T]): T = { | ||
| try { | ||
| Await.result(awaitable, duration) | ||
| } catch addMessageIfTimeout | ||
| } | ||
| } | ||
|
|
||
|
|
||
| private[spark] object RpcTimeout { | ||
|
|
||
| /** | ||
| * Lookup the timeout property in the configuration and create | ||
| * a RpcTimeout with the property key in the description. | ||
| * @param conf configuration properties containing the timeout | ||
| * @param timeoutProp property key for the timeout in seconds | ||
| * @throws NoSuchElementException if property is not set | ||
| */ | ||
| def apply(conf: SparkConf, timeoutProp: String): RpcTimeout = { | ||
| val timeout = { conf.getTimeAsSeconds(timeoutProp) seconds } | ||
| new RpcTimeout(timeout, timeoutProp) | ||
| } | ||
|
|
||
| /** | ||
| * Lookup the timeout property in the configuration and create | ||
| * a RpcTimeout with the property key in the description. | ||
| * Uses the given default value if property is not set | ||
| * @param conf configuration properties containing the timeout | ||
| * @param timeoutProp property key for the timeout in seconds | ||
| * @param defaultValue default timeout value in seconds if property not found | ||
| */ | ||
| def apply(conf: SparkConf, timeoutProp: String, defaultValue: String): RpcTimeout = { | ||
| val timeout = { conf.getTimeAsSeconds(timeoutProp, defaultValue) seconds } | ||
|
Member
Author
There was a problem hiding this comment. Should this be
Contributor
There was a problem hiding this comment. as mentioned above, I don't think so -- lets stick w/ the current behavior of using |
||
| new RpcTimeout(timeout, timeoutProp) | ||
| } | ||
|
|
||
| /** | ||
| * Lookup prioritized list of timeout properties in the configuration | ||
| * and create a RpcTimeout with the first set property key in the | ||
| * description. | ||
| * Uses the given default value if property is not set | ||
| * @param conf configuration properties containing the timeout | ||
| * @param timeoutPropList prioritized list of property keys for the timeout in seconds | ||
| * @param defaultValue default timeout value in seconds if no properties found | ||
| */ | ||
| def apply(conf: SparkConf, timeoutPropList: Seq[String], defaultValue: String): RpcTimeout = { | ||
|
Member
Author
There was a problem hiding this comment. Creating |
||
| require(timeoutPropList.nonEmpty) | ||
|
|
||
| // Find the first set property or use the default value with the first property | ||
| val itr = timeoutPropList.iterator | ||
| var foundProp: Option[(String, String)] = None | ||
| while (itr.hasNext && foundProp.isEmpty){ | ||
| val propKey = itr.next() | ||
| conf.getOption(propKey).foreach { prop => foundProp = Some(propKey, prop) } | ||
| } | ||
| val finalProp = foundProp.getOrElse(timeoutPropList.head, defaultValue) | ||
|
Member
There was a problem hiding this comment.
Member
There was a problem hiding this comment. Never mind. Looks it's long too. |
||
| val timeout = { Utils.timeStringAsSeconds(finalProp._2) seconds } | ||
| new RpcTimeout(timeout, finalProp._1) | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: you don't need to import
FiniteDurationexplicitly since you're importingduration._. Also you are supposed to order direct class imports before package imports (not just alphabetically), so it should be:the intellij import organizer will get this wrong, but aaron davidson wrote a plugin which does it right -- there are instructions for using it here: https://cwiki.apache.org/confluence/display/SPARK/Spark+Code+Style+Guide#SparkCodeStyleGuide-Imports
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks, using the plugin now