[GitHub] spark pull request #16690: [SPARK-19347] ReceiverSupervisorImpl can add bloc...
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/16690#discussion_r101616883
--- Diff: core/src/main/scala/org/apache/spark/rpc/RpcEndpointRef.scala ---
@@ -75,10 +105,11 @@ private[spark] abstract class RpcEndpointRef(conf:
SparkConf)
* @tparam T type of the reply message
* @return the reply message from the corresponding [[RpcEndpoint]]
*/
+ @deprecated("use 'askSync' instead.", "2.2.0")
--- End diff --
It seems like this has caused the build to produce a lot of deprecation
warnings. @jinxing64 could the callers of this method be changed, in Spark, to
use the new alternative?
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16690: [SPARK-19347] ReceiverSupervisorImpl can add bloc...
Github user asfgit closed the pull request at: https://github.com/apache/spark/pull/16690 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16690: [SPARK-19347] ReceiverSupervisorImpl can add bloc...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/16690#discussion_r98747972
--- Diff: core/src/main/scala/org/apache/spark/rpc/RpcEndpointRef.scala ---
@@ -75,10 +106,11 @@ private[spark] abstract class RpcEndpointRef(conf:
SparkConf)
* @tparam T type of the reply message
* @return the reply message from the corresponding [[RpcEndpoint]]
*/
+ @deprecated("use 'askSync' instead.", "2.1.0")
--- End diff --
2.2.0
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16690: [SPARK-19347] ReceiverSupervisorImpl can add bloc...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/16690#discussion_r98748039 --- Diff: core/src/main/scala/org/apache/spark/rpc/RpcEndpointRef.scala --- @@ -19,6 +19,7 @@ package org.apache.spark.rpc import scala.concurrent.Future import scala.reflect.ClassTag +import scala.util.control.NonFatal --- End diff -- not used --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16690: [SPARK-19347] ReceiverSupervisorImpl can add bloc...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/16690#discussion_r98747902
--- Diff: core/src/main/scala/org/apache/spark/rpc/RpcEndpointRef.scala ---
@@ -91,6 +123,7 @@ private[spark] abstract class RpcEndpointRef(conf:
SparkConf)
* @tparam T type of the reply message
* @return the reply message from the corresponding [[RpcEndpoint]]
*/
+ @deprecated("use 'askSync' instead.", "2.1.0")
--- End diff --
2.2.0
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16690: [SPARK-19347] ReceiverSupervisorImpl can add bloc...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/16690#discussion_r98297004
--- Diff: core/src/main/scala/org/apache/spark/rpc/RpcEndpointRef.scala ---
@@ -19,11 +19,13 @@ package org.apache.spark.rpc
import scala.concurrent.Future
import scala.reflect.ClassTag
+import scala.util.control.NonFatal
import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.internal.Logging
import org.apache.spark.util.RpcUtils
+
--- End diff --
nit: don't add.
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16690: [SPARK-19347] ReceiverSupervisorImpl can add bloc...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/16690#discussion_r98297290
--- Diff: core/src/main/scala/org/apache/spark/rpc/RpcEndpointRef.scala ---
@@ -63,8 +65,48 @@ private[spark] abstract class RpcEndpointRef(conf:
SparkConf)
def ask[T: ClassTag](message: Any): Future[T] = ask(message,
defaultAskTimeout)
/**
- * Send a message to the corresponding [[RpcEndpoint]] and get its
result within a default
- * timeout, or throw a SparkException if this fails even after the
default number of retries.
+ * Send a message to the corresponding [[RpcEndpoint.receiveAndReply]]
and get its result within a
+ * default timeout, throw a SparkException if this fails.
+ *
+ * Note: this is a blocking action which may cost a lot of time, so
don't call it in a message
+ * loop of [[RpcEndpoint]].
+
+ * @param message the message to send
+ * @tparam T type of the reply message
+ * @return the reply message from the corresponding [[RpcEndpoint]]
+ */
+ def askWithBlocking[T: ClassTag](message: Any): T =
askWithBlocking(message, defaultAskTimeout)
+
+ /**
+ * Send a message to the corresponding [[RpcEndpoint.receiveAndReply]]
and get its result within a
+ * specified timeout, throw a SparkException if this fails.
+ *
+ * Note: this is a blocking action which may cost a lot of time, so
don't call it in a message
+ * loop of [[RpcEndpoint]].
+ *
+ * @param message the message to send
+ * @param timeout the timeout duration
+ * @tparam T type of the reply message
+ * @return the reply message from the corresponding [[RpcEndpoint]]
+ */
+ def askWithBlocking[T: ClassTag](message: Any, timeout: RpcTimeout): T =
{
+try {
+ val future = ask[T](message, timeout)
+ val result = timeout.awaitResult(future)
+ if (result == null) {
--- End diff --
This is not an error. It's perfectly legitimate to return null.
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16690: [SPARK-19347] ReceiverSupervisorImpl can add bloc...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/16690#discussion_r98297392
--- Diff: core/src/main/scala/org/apache/spark/rpc/RpcEndpointRef.scala ---
@@ -63,8 +65,48 @@ private[spark] abstract class RpcEndpointRef(conf:
SparkConf)
def ask[T: ClassTag](message: Any): Future[T] = ask(message,
defaultAskTimeout)
/**
- * Send a message to the corresponding [[RpcEndpoint]] and get its
result within a default
- * timeout, or throw a SparkException if this fails even after the
default number of retries.
+ * Send a message to the corresponding [[RpcEndpoint.receiveAndReply]]
and get its result within a
+ * default timeout, throw a SparkException if this fails.
+ *
+ * Note: this is a blocking action which may cost a lot of time, so
don't call it in a message
+ * loop of [[RpcEndpoint]].
+
+ * @param message the message to send
+ * @tparam T type of the reply message
+ * @return the reply message from the corresponding [[RpcEndpoint]]
+ */
+ def askWithBlocking[T: ClassTag](message: Any): T =
askWithBlocking(message, defaultAskTimeout)
+
+ /**
+ * Send a message to the corresponding [[RpcEndpoint.receiveAndReply]]
and get its result within a
+ * specified timeout, throw a SparkException if this fails.
+ *
+ * Note: this is a blocking action which may cost a lot of time, so
don't call it in a message
+ * loop of [[RpcEndpoint]].
+ *
+ * @param message the message to send
+ * @param timeout the timeout duration
+ * @tparam T type of the reply message
+ * @return the reply message from the corresponding [[RpcEndpoint]]
+ */
+ def askWithBlocking[T: ClassTag](message: Any, timeout: RpcTimeout): T =
{
+try {
+ val future = ask[T](message, timeout)
+ val result = timeout.awaitResult(future)
+ if (result == null) {
+throw new SparkException("RpcEndpoint returned null")
+ }
+ return result
+} catch {
+ case NonFatal(e) =>
+throw new SparkException(
--- End diff --
Isn't it better to just propagate the original exception? You can get the
context from the stack trace.
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16690: [SPARK-19347] ReceiverSupervisorImpl can add bloc...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/16690#discussion_r98297209 --- Diff: core/src/main/scala/org/apache/spark/rpc/RpcEndpointRef.scala --- @@ -63,8 +65,48 @@ private[spark] abstract class RpcEndpointRef(conf: SparkConf) def ask[T: ClassTag](message: Any): Future[T] = ask(message, defaultAskTimeout) /** - * Send a message to the corresponding [[RpcEndpoint]] and get its result within a default - * timeout, or throw a SparkException if this fails even after the default number of retries. + * Send a message to the corresponding [[RpcEndpoint.receiveAndReply]] and get its result within a + * default timeout, throw a SparkException if this fails. + * + * Note: this is a blocking action which may cost a lot of time, so don't call it in a message + * loop of [[RpcEndpoint]]. + + * @param message the message to send + * @tparam T type of the reply message + * @return the reply message from the corresponding [[RpcEndpoint]] + */ + def askWithBlocking[T: ClassTag](message: Any): T = askWithBlocking(message, defaultAskTimeout) --- End diff -- `askWithBlocking` is a weird name. I'd use `blockingAsk`, or `askSync`. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #16690: [SPARK-19347] ReceiverSupervisorImpl can add bloc...
GitHub user jinxing64 opened a pull request:
https://github.com/apache/spark/pull/16690
[SPARK-19347] ReceiverSupervisorImpl can add block to ReceiverTracker
multiple times because of askWithRetry.
## What changes were proposed in this pull request?
`ReceiverSupervisorImpl` on executor side reports block's meta back to
`ReceiverTracker` on driver side. In current code, `askWithRetry` is used.
However, for `AddBlock`, `ReceiverTracker` is not idempotent, which may result
in messages are processed multiple times.
*To reproduce*:
1. Check if it is the first time receiving `AddBlock` in `ReceiverTracker`,
if so sleep long enough(say 200 seconds), thus the first RPC call will be
timeout in `askWithRetry`, then `AddBlock` will be resent.
2. Rebuild Spark and run following job:
```
def streamProcessing(): Unit = {
val conf = new SparkConf()
.setAppName("StreamingTest")
.setMaster(masterUrl)
val ssc = new StreamingContext(conf, Seconds(200))
val stream = ssc.socketTextStream("localhost", 1234)
stream.print()
ssc.start()
ssc.awaitTermination()
}
```
To fix:
It makes sense to provide a blocking version `ask` in RpcEndpointRef, as
mentioned in SPARK-18113
(https://github.com/apache/spark/pull/16503#event-927953218). Because Netty RPC
layer will not drop messages. `askWithRetry` is a leftover from akka days. It
imposes restrictions on the caller(e.g. idempotency) and other things that
people generally don't pay that much attention to when using it.
## How was this patch tested?
Test manually. The scenario described above doesn't happen with this patch.
Please review http://spark.apache.org/contributing.html before opening a
pull request.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/jinxing64/spark SPARK-19347
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/16690.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #16690
commit c5bcccf227446f5d044f8fb0518caa12cfef7421
Author: jinxing
Date: 2017-01-24T09:33:23Z
[SPARK-19347] ReceiverSupervisorImpl can add block to ReceiverTracker
multiple times because of askWithRetry
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
