[SPARK-18113] Use ask to replace askWithRetry in canCommit and make receiver idempotent. #16503

jinxing64 · 2017-01-08T16:26:20Z

What changes were proposed in this pull request?

Method canCommit sends AskPermissionToCommitOutput using askWithRetry. If timeout, it will send again. Thus AskPermissionToCommitOutput can be received multi times. Method canCommit should return the same value when called by the same attempt multi times.

In implementation before this fix, method handleAskPermissionToCommit just check if there is committer already registered, which is not enough. When worker retries AskPermissionToCommitOutput it will get CommitDeniedException, then the task will fail with reason TaskCommitDenied, which is not regarded as a task failure(SPARK-11178), so TaskScheduler will schedule this task infinitely.

In this fix, use ask to replace askWithRetry in canCommit and make receiver idempotent.

How was this patch tested?

Added a new unit test to OutputCommitCoordinatorSuite.

jinxing64 · 2017-01-09T02:52:06Z

@mccheah @JoshRosen @ash211 Could you please take look at this?

jinxing64 · 2017-01-09T15:38:13Z

@zsxwing @kayousterhout @andrewor14 Could you please help take a look at this ?

vanzin · 2017-01-09T20:01:37Z

I think this is another case where using askWithRetry makes no sense given the guarantees of the RPC layer.

zsxwing · 2017-01-09T22:47:30Z

Good catch. Looks good to me.

@vanzin The RPC layer only guarantees at-most-once. Retry may be still helpful in some case, but the receiver should be idempotent. Either the current change or changing to use ask (Spark will retry the task) is good to me.

vanzin · 2017-01-09T22:52:08Z

The RPC layer only guarantees at-most-once

That was the case with akka (I think, not really sure), but the netty RPC layer doesn't drop messages. The new one is "exactly once".

zsxwing · 2017-01-09T22:54:56Z

That was the case with akka (I think, not really sure), but the netty RPC layer doesn't drop messages. The new one is "exactly once".

It doesn't drop but the connection may be broken. askWithRetry will reconnect if the connection is broken.

vanzin · 2017-01-09T22:57:03Z

It doesn't drop but the connection may be broken

In which case the executor will die (see CoarseGrainedExecutorBackend::onDisconnected).

zsxwing · 2017-01-09T23:00:55Z

In which case the executor will die (see CoarseGrainedExecutorBackend::onDisconnected).

Yeah. Didn't recall that. Then I agree that using ask is better.

jinxing64 · 2017-01-10T02:38:51Z

@zsxing, @vanzin
Maybe using ask in method canCommit is not suitable(i think). Because ask returns a Future, but it should be a blocking process to get result of AskPermissionToCommitOutput in canCommit. RPC layer doesn't drop message but message can be timeout. If timeout happens, we can retry or fail the task and let spark reschedule it, both of them are ok. But for some heavy tasks, retry is light weight (i think). So my suggestion is to keep the askWithRetry here and change the receiver to be idempotent. What do you think ?

vanzin · 2017-01-10T17:40:29Z

You can make ask blocking by waiting for its future (e.g. with ThreadUtils.awaitResult).

My point of not using askWithRetry is that it's basically an unneeded API, and a leftover from the akka days that doesn't make sense anymore. It's prone to cause deadlocks (exactly because it's blocking), 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.

If we can remove uses of askWithRetry as we find these issues, we can, at some point, finally get rid of the API altogether.

RPC layer doesn't drop message but message can be timeout.

Yes it can timeout. You can retry it (basically doing what askWithRetry does) but it should be such an edge case that failing the task should be ok.

If you think about how the RPC layer works when you use askWithRetry, this is what happens:

first RPC is sent
remote end is blocked on something, RPC is waiting in the queue
sender re-sends the RPC
lather, rinse, repeat
at some point, receive goes through the RPC queue and start responding to the RPCs
it responds to the first RPC above first, sender ignores the answer since RPC was timed out
lather, rinse, repeat
finally the last RPC is responded to and the sender sees the reply

So it's a really expensive way of just doing ask with a longer timeout.

jinxing64 · 2017-01-11T04:48:25Z

@vanzin
Thanks a lot for your comment. It's very helpful.
I'll change it to ask.
I think it make sense to keep receiver idempotent when handling AskPermissionToCommitOutput, even though we use ask to replace askWithRetry. So I didn't remove the code and unit test.

jinxing64 · 2017-01-11T04:58:43Z

If we can remove uses of askWithRetry as we find these issues, we can, at some point, finally get rid of the API altogether.

How do you think about providing a "blocking" ask in RpcEndpointRef? Just wait the future but no retry. Thus no need to wait the future outside of RpcEndpontRef. If you agree, I'll make another PR for changing. Currently I already found some places using a blocking ask, that's why I think it's needed to provid one in RpcEndpointRef.

jinxing64 · 2017-01-12T05:43:07Z

ping @zsxwing @vanzin
Could you give another look at this please ?

ash211 · 2017-01-11T20:01:59Z

core/src/test/scala/org/apache/spark/scheduler/OutputCommitCoordinatorSuite.scala

@@ -189,6 +188,13 @@ class OutputCommitCoordinatorSuite extends SparkFunSuite with BeforeAndAfter {
    assert(
      !outputCommitCoordinator.canCommit(stage, partition, nonAuthorizedCommitter + 3))
  }
+
+  test("Authoried commiter get true if it calls canCommit again.") {


Reword test description to: Duplicate calls to canCommit from the authorized committer gets idempotent responses

ash211 · 2017-01-11T20:04:41Z

core/src/test/scala/org/apache/spark/scheduler/OutputCommitCoordinatorSuite.scala

@@ -221,6 +227,22 @@ private case class OutputCommitFunctions(tempDirPath: String) {
      if (ctx.attemptNumber == 0) failingOutputCommitter else successfulOutputCommitter)
  }

+  // Receiver should be idempotent for AskPermissionToCommitOutput
+  def callCanCommitMultiTimes(iter: Iterator[Int]): Unit = {


nit rename to callCanCommitMultipleTimes

ash211 · 2017-01-11T20:06:23Z

core/src/main/scala/org/apache/spark/scheduler/OutputCommitCoordinator.scala

-            false
+            // Coordinator should be idempotent when receiving AskPermissionToCommit.
+            if (existingCommitter == attemptNumber) {
+              true


please log a warning here before returning true -- reaching this branch is likely indicative of network problems

logWarning(s"Authorizing duplicate request to commit for " + s"attemptNumber=$attemptNumber to commit for stage=$stage, partition=$partition; " + s"existingCommitter = $existingCommitter. This can indicate dropped network traffic.")

jinxing64 · 2017-01-12T09:12:11Z

@ash211
Thank you so much for your comment. I've changed accordingly.
Could you please give another look?

ash211 · 2017-01-12T10:21:03Z

You covered my concerns!

I think this will fix some parts of this problem for sure, not sure if it covers every possible case though.

vanzin · 2017-01-12T18:52:45Z

ok to test

SparkQA · 2017-01-12T18:55:52Z

Test build #71274 has finished for PR 16503 at commit 6c25755.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

ash211 · 2017-01-12T19:14:07Z

@jinxing64 can you please fix the failing Scala style tests? You can run them locally with ./dev/scalastyle

jinxing64 · 2017-01-13T03:11:12Z

@ash211
Thanks a lot for your comment. I've already fixed the failing Scala style tests. Running ./dev/scalastyle passed. Could you give another look?

SparkQA · 2017-01-13T05:35:08Z

Test build #71284 has finished for PR 16503 at commit aba406d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2017-01-13T20:10:52Z

Looks good. If you want to add a blocking version of ask (and add a deprecation tag to askWithRetry) that would be fine too. But maybe a separate change.

jinxing64 · 2017-01-13T23:20:17Z

@vanzin @zsxwing
Thanks a lot for your comment. I will file another jira to add a blocking version of ask.
What else can I do for this pr : ) ?

vanzin · 2017-01-14T00:26:56Z

core/src/test/scala/org/apache/spark/scheduler/OutputCommitCoordinatorSuite.scala

+      .canCommit(ctx.stageId(), ctx.partitionId(), ctx.attemptNumber())
+    val canCommit2 = SparkEnv.get.outputCommitCoordinator
+      .canCommit(ctx.stageId(), ctx.partitionId(), ctx.attemptNumber())
+    if(canCommit1 && canCommit2) {


nit: space after if

vanzin · 2017-01-14T00:27:26Z

core/src/test/scala/org/apache/spark/scheduler/OutputCommitCoordinatorSuite.scala

+    val canCommit2 = SparkEnv.get.outputCommitCoordinator
+      .canCommit(ctx.stageId(), ctx.partitionId(), ctx.attemptNumber())
+    if(canCommit1 && canCommit2) {
+      Utils.createDirectory(tempDirPath)


Do you need this? Can you just assert that you got permission both times (and rely on the exception causing the runJob call to fail in the caller)?

Yes, using assert is better here.

ash211 · 2017-01-14T00:58:05Z

Making this idempotent looks great. I think there's a separate issue with this code still not handling poorly-timed preemption, but let's deal with that in a separate ticket / PR.

Good work so far @jinxing64 !

SparkQA · 2017-01-14T19:38:55Z

Test build #71378 has finished for PR 16503 at commit eb5367a.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-01-14T19:47:23Z

Test build #71377 has finished for PR 16503 at commit b867b92.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-01-14T20:06:13Z

Test build #71380 has finished for PR 16503 at commit 55e4fd3.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

jinxing64 · 2017-01-15T00:22:26Z

@vanzin @ash211
Thanks a lot for your comments; I've changed accordingly. Please give another look at this~~

jinxing64 · 2017-01-16T15:48:33Z

ping

jinxing64 · 2017-01-18T01:03:53Z

@vanzin
Thanks for your comments.I have changed the unit test. Could you take another look?

…eceiver idempotent.

SparkQA · 2017-01-18T05:06:50Z

Test build #71560 has finished for PR 16503 at commit 69b412a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-01-18T05:06:53Z

Test build #71558 has finished for PR 16503 at commit 52af8c5.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

jinxing64 · 2017-01-18T05:12:35Z

@vanzin
Sorry for the stupid mistake I made. I've changed. Please take another look.

vanzin · 2017-01-18T18:46:57Z

Merging to master.

…eceiver idempotent. ## What changes were proposed in this pull request? Method canCommit sends AskPermissionToCommitOutput using askWithRetry. If timeout, it will send again. Thus AskPermissionToCommitOutput can be received multi times. Method canCommit should return the same value when called by the same attempt multi times. In implementation before this fix, method handleAskPermissionToCommit just check if there is committer already registered, which is not enough. When worker retries AskPermissionToCommitOutput it will get CommitDeniedException, then the task will fail with reason TaskCommitDenied, which is not regarded as a task failure(SPARK-11178), so TaskScheduler will schedule this task infinitely. In this fix, use `ask` to replace `askWithRetry` in `canCommit` and make receiver idempotent. ## How was this patch tested? Added a new unit test to OutputCommitCoordinatorSuite. Author: jinxing <[email protected]> Closes apache#16503 from jinxing64/SPARK-18113.

… 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 (apache#16503 (comment)). 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. Author: jinxing <[email protected]> Closes apache#16690 from jinxing64/SPARK-19347.

…eceiver idempotent. ## What changes were proposed in this pull request? Method canCommit sends AskPermissionToCommitOutput using askWithRetry. If timeout, it will send again. Thus AskPermissionToCommitOutput can be received multi times. Method canCommit should return the same value when called by the same attempt multi times. In implementation before this fix, method handleAskPermissionToCommit just check if there is committer already registered, which is not enough. When worker retries AskPermissionToCommitOutput it will get CommitDeniedException, then the task will fail with reason TaskCommitDenied, which is not regarded as a task failure(SPARK-11178), so TaskScheduler will schedule this task infinitely. In this fix, use `ask` to replace `askWithRetry` in `canCommit` and make receiver idempotent. ## How was this patch tested? Added a new unit test to OutputCommitCoordinatorSuite. Author: jinxing <[email protected]> Closes apache#16503 from jinxing64/SPARK-18113.

… 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 (apache#16503 (comment)). 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. Author: jinxing <[email protected]> Closes apache#16690 from jinxing64/SPARK-19347.

## What changes were proposed in this pull request? `askSync` is already added in `RpcEndpointRef` (see SPARK-19347 and apache#16690 (comment)) and `askWithRetry` is marked as deprecated. As mentioned SPARK-18113(apache#16503 (comment)): >askWithRetry is basically an unneeded API, and a leftover from the akka days that doesn't make sense anymore. It's prone to cause deadlocks (exactly because it's blocking), 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. Since `askWithRetry` is just used inside spark and not in user logic. It might make sense to replace all of them with `askSync`. ## How was this patch tested? This PR doesn't change code logic, existing unit test can cover. Author: jinxing <[email protected]> Closes apache#16790 from jinxing64/SPARK-19450.

…eceiver idempotent. ## What changes were proposed in this pull request? Method canCommit sends AskPermissionToCommitOutput using askWithRetry. If timeout, it will send again. Thus AskPermissionToCommitOutput can be received multi times. Method canCommit should return the same value when called by the same attempt multi times. In implementation before this fix, method handleAskPermissionToCommit just check if there is committer already registered, which is not enough. When worker retries AskPermissionToCommitOutput it will get CommitDeniedException, then the task will fail with reason TaskCommitDenied, which is not regarded as a task failure(SPARK-11178), so TaskScheduler will schedule this task infinitely. In this fix, use `ask` to replace `askWithRetry` in `canCommit` and make receiver idempotent. ## How was this patch tested? Added a new unit test to OutputCommitCoordinatorSuite. Author: jinxing <[email protected]> Closes apache#16503 from jinxing64/SPARK-18113.

@pr

…eceiver idempotent.@pr[apache#16503]

…eceiver idempotent. ## What changes were proposed in this pull request? Method canCommit sends AskPermissionToCommitOutput using askWithRetry. If timeout, it will send again. Thus AskPermissionToCommitOutput can be received multi times. Method canCommit should return the same value when called by the same attempt multi times. In implementation before this fix, method handleAskPermissionToCommit just check if there is committer already registered, which is not enough. When worker retries AskPermissionToCommitOutput it will get CommitDeniedException, then the task will fail with reason TaskCommitDenied, which is not regarded as a task failure(SPARK-11178), so TaskScheduler will schedule this task infinitely. In this fix, use `ask` to replace `askWithRetry` in `canCommit` and make receiver idempotent. ## How was this patch tested? Added a new unit test to OutputCommitCoordinatorSuite. Author: jinxing <[email protected]> Closes apache#16503 from jinxing64/SPARK-18113.

jinxing64 changed the title ~~[SPARK-18113] Method canCommit should return the same value when call…~~ [SPARK-18113] canCommit should return same when called by same attempt multi times. Jan 8, 2017

jinxing64 force-pushed the SPARK-18113 branch from dc188f4 to 06760ea Compare January 11, 2017 04:43

jinxing64 changed the title ~~[SPARK-18113] canCommit should return same when called by same attempt multi times.~~ [SPARK-18113] Use ask to replace askWithRetry in canCommit and make receiver idempotent. Jan 11, 2017

ash211 suggested changes Jan 12, 2017

View reviewed changes

jinxing64 force-pushed the SPARK-18113 branch from 6c25755 to aba406d Compare January 13, 2017 03:08

vanzin reviewed Jan 14, 2017

View reviewed changes

jinxing64 force-pushed the SPARK-18113 branch from aba406d to 26c9a2f Compare January 14, 2017 16:36

jinxing64 force-pushed the SPARK-18113 branch from 55e4fd3 to 52af8c5 Compare January 18, 2017 02:32

[SPARK-18113] Use ask to replace askWithRetry in canCommit and make r…

69b412a

…eceiver idempotent.

jinxing64 force-pushed the SPARK-18113 branch from 52af8c5 to 69b412a Compare January 18, 2017 02:33

asfgit closed this in 33791a8 Jan 18, 2017

jinxing64 mentioned this pull request Jan 24, 2017

[SPARK-19347] ReceiverSupervisorImpl can add block to ReceiverTracker multiple times because of askWithRetry. #16690

Closed

jinxing64 mentioned this pull request Feb 3, 2017

[SPARK-19450] Replace askWithRetry with askSync. #16790

Closed

vanzin mentioned this pull request May 5, 2017

[SPARK-20564][Deploy] Reduce massive executor failures when executor count is large (>2000) #17854

Closed

IceMimosa pushed a commit to IceMimosa/spark that referenced this pull request May 8, 2018

[SPARK-18113] Use ask to replace askWithRetry in canCommit and make r…

6419db7

…eceiver idempotent.@pr[apache#16503]

IceMimosa mentioned this pull request May 8, 2018

[SPARK-18113] Use ask to replace askWithRetry in canCommit and make r… growingio/spark#3

Merged

[SPARK-18113] Use ask to replace askWithRetry in canCommit and make receiver idempotent. #16503

[SPARK-18113] Use ask to replace askWithRetry in canCommit and make receiver idempotent. #16503

Conversation

jinxing64 commented Jan 8, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

jinxing64 commented Jan 9, 2017

jinxing64 commented Jan 9, 2017

vanzin commented Jan 9, 2017

zsxwing commented Jan 9, 2017

vanzin commented Jan 9, 2017

zsxwing commented Jan 9, 2017

vanzin commented Jan 9, 2017

zsxwing commented Jan 9, 2017

jinxing64 commented Jan 10, 2017

vanzin commented Jan 10, 2017

jinxing64 commented Jan 11, 2017

jinxing64 commented Jan 11, 2017

jinxing64 commented Jan 12, 2017

ash211 Jan 11, 2017

Choose a reason for hiding this comment

ash211 Jan 11, 2017

Choose a reason for hiding this comment

ash211 Jan 11, 2017

Choose a reason for hiding this comment

jinxing64 commented Jan 12, 2017

ash211 commented Jan 12, 2017

vanzin commented Jan 12, 2017

SparkQA commented Jan 12, 2017

ash211 commented Jan 12, 2017 • edited Loading

jinxing64 commented Jan 13, 2017

SparkQA commented Jan 13, 2017

vanzin commented Jan 13, 2017

jinxing64 commented Jan 13, 2017

vanzin Jan 14, 2017

Choose a reason for hiding this comment

vanzin Jan 14, 2017

Choose a reason for hiding this comment

jinxing64 Jan 15, 2017

Choose a reason for hiding this comment

ash211 commented Jan 14, 2017

SparkQA commented Jan 14, 2017

SparkQA commented Jan 14, 2017

SparkQA commented Jan 14, 2017

jinxing64 commented Jan 15, 2017

jinxing64 commented Jan 16, 2017

jinxing64 commented Jan 18, 2017

SparkQA commented Jan 18, 2017

SparkQA commented Jan 18, 2017

jinxing64 commented Jan 18, 2017

vanzin commented Jan 18, 2017

jinxing64 commented Jan 8, 2017 •

edited

Loading

ash211 commented Jan 12, 2017 •

edited

Loading