Github user BryanCutler commented on a diff in the pull request:
https://github.com/apache/spark/pull/6205#discussion_r30625296
--- Diff: core/src/main/scala/org/apache/spark/rpc/RpcEnv.scala ---
@@ -182,3 +185,89 @@ private[spark] object RpcAddress {
RpcAddress(host, port)
}
}
+
+
+/**
+ * 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 timeout timeout duration in seconds
+ * @param description description to be displayed in a timeout exception
+ */
+private[spark] class RpcTimeout(timeout: FiniteDuration, description:
String) {
+
+ /** Get the timeout duration */
+ def duration: FiniteDuration = timeout
+
+ /** Get the message associated with this timeout */
+ def message: String = description
+
+ /** Amends the standard message of TimeoutException to include the
description */
+ def amend(te: TimeoutException): TimeoutException = {
+ new TimeoutException(te.getMessage() + " " + description)
+ }
+
+ /** Wait on a future result to catch and amend a TimeoutException */
+ def awaitResult[T](future: Future[T]): T = {
+ try {
+ Await.result(future, duration)
+ }
+ catch {
+ case te: TimeoutException => throw amend(te)
+ }
+ }
+}
+
+
+/**
+ * Create an RpcTimeout using a configuration property that controls the
timeout duration so when
+ * a TimeoutException is thrown, the property key will be indicated in the
message.
+ */
+object RpcTimeout {
+
+ private[this] val messagePrefix = "This timeout is controlled by "
+
+ /**
+ * 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, messagePrefix + 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 }
+ new RpcTimeout(timeout, messagePrefix + 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 = {
--- End diff --
Creating `RpcTimeout` from a prioritized list of property keys, from
@squito previous request
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
