[SPARK-4873][Streaming] Use `Future.zip` instead of `Future.flatMap`(for-loop) in WriteAheadLogBasedBlockHandler #3721

zsxwing · 2014-12-17T09:56:29Z

Use Future.zip instead of Future.flatMap(for-loop). zip implies these two Futures will run concurrently, while flatMap usually means one Future depends on the other one.

SparkQA · 2014-12-17T09:57:28Z

Test build #24542 has started for PR 3721 at commit 860f3a2.

This patch merges cleanly.

SparkQA · 2014-12-17T10:22:27Z

Test build #24543 has started for PR 3721 at commit 8b6d095.

This patch merges cleanly.

SparkQA · 2014-12-17T11:20:53Z

Test build #24542 has finished for PR 3721 at commit 860f3a2.

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

AmplabJenkins · 2014-12-17T11:20:57Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24542/
Test PASSed.

SparkQA · 2014-12-17T11:47:12Z

Test build #24543 has finished for PR 3721 at commit 8b6d095.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- trait ParquetTest
- protected class CaseInsensitiveMap(map: Map[String, String]) extends Map[String, String]

AmplabJenkins · 2014-12-17T11:47:15Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24543/
Test PASSed.

JoshRosen · 2014-12-17T20:56:24Z

The flaky test fix here looks like it addresses https://issues.apache.org/jira/browse/SPARK-4790.

/cc @harishreedharan, who is investigating that flaky test.

harishreedharan · 2014-12-17T21:30:01Z

This would not address the test issue as we are still returning from cleanupOldLogs before the deletion is completed. This tracker3.cleanupOldBatches(batchTime2) will still end up hitting an exception if the deletion is not done in between the listFiles and getFileStatus (for the child file) calls in the getFileStatus method.

harishreedharan · 2014-12-17T21:31:13Z

The eventually ensures that the block is retried - but the failure is happening before the eventually, so the test would still throw.

harishreedharan · 2014-12-17T21:45:04Z

Hmm, looking at it again - this would fix the test as well, though I think the approach in #3726 is cleaner.

zsxwing · 2014-12-18T02:02:07Z

My first trial was making cleanupOldBatches return a Future so that the caller can use it to wait. But I looked other places and found they used eventually. So I followed their approach. But IMO it's better to return the Future to user.

zsxwing · 2014-12-18T02:45:37Z

Used Future to fix the test and override my previous commits.

SparkQA · 2014-12-18T02:47:28Z

Test build #24566 has started for PR 3721 at commit d3d8a51.

This patch merges cleanly.

SparkQA · 2014-12-18T04:11:48Z

Test build #24566 has finished for PR 3721 at commit d3d8a51.

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

AmplabJenkins · 2014-12-18T04:11:52Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24566/
Test PASSed.

harishreedharan · 2014-12-18T07:50:28Z

streaming/src/main/scala/org/apache/spark/streaming/util/WriteAheadLogManager.scala

@@ -125,7 +125,7 @@ private[streaming] class WriteAheadLogManager(
   * between the node calculating the threshTime (say, driver node), and the local system time
   * (say, worker node), the caller has to take account of possible time skew.
   */
-  def cleanupOldLogs(threshTime: Long): Unit = {
+  def cleanupOldLogs(threshTime: Long): Future[Unit] = {


I don't think we should be exposing this future. That is internal implementation detail and we'd have to stick with this implementation. At some point, we might want to delete the files synchronously - at which point returning this Future might not make sense.

IMO, I prefer to also return Future for an asynchronous action. Returning Unit hides the asynchronous feature and such method will be misused easily.

The thing is that deleting asynchronously is an implementation detail - which we choose to do now, which can be changed later, don't you think? If we change it later, we'd have to change this method's signature - which can cause pain if there is code that uses Await.result(*) on this future.

If we expose it via a parameter, we can choose to ignore the param and still the calling code will not have to change. Since it is unlikely that the calling code will actually depend on the async nature, it is unlikely to see any difference in functionality and no change in code is required.

The thing is that deleting asynchronously is an implementation detail - which we choose to do now, which can be changed later, don't you think? If we change it later, we'd have to change this method's signature - which can cause pain if there is code that uses Await.result(*) on this future.

I doubt if it will be changed. However, asynchronously is an important implementation detail that the caller should know it, or they may misuse it.

If we expose it via a parameter, we can choose to ignore the param and still the calling code will not have to change. Since it is unlikely that the calling code will actually depend on the async nature, it is unlikely to see any difference in functionality and no change in code is required.

I don't think a parameter is enough. At least, it needs a more parameter, a timeout parameter. In your PR, you used 1 second which may not be enough.

What's more, if we really want to change it to a synchronously deleting. Returning Future does still work. Just simply writing something like:

deleteFiles() return Promise[Unit]().success(null).future

tdas · 2014-12-24T01:39:13Z

streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceivedBlockHandler.scala

@@ -169,10 +171,7 @@ private[streaming] class WriteAheadLogBasedBlockHandler(
    }

    // Combine the futures, wait for both to complete, and return the write ahead log segment
-    val combinedFuture = for {


Very good catch. I wasnt aware that for-yield is flatMap which is obviously ordered.

tdas · 2014-12-24T22:05:49Z

High level comment, I think I completely agree on the Future.zip, but not sure if I find the other changed related to other two valid.

There is already a separate execution context implicitly available.
The flaky test, I think the solution in [SPARK-4790][STREAMING] Fix ReceivedBlockTrackerSuite waits for old file... #3726 is closer to the ideal as changing the signature to return Future exposes unnecessary internal details.

So it would be great if you can reduce the scope of this PR to the Future.zip and accordingly update the title or the PR + JIRA to a more specific than "improvements".

zsxwing · 2014-12-25T01:46:19Z

Now this PR only contains Future.zip change. I also updated the PR & JIRA.

SparkQA · 2014-12-25T01:47:34Z

Test build #24799 has started for PR 3721 at commit 46a2cd9.

This patch merges cleanly.

tdas · 2014-12-25T01:49:08Z

LGTM. Will merge if tests pass.

SparkQA · 2014-12-25T03:12:55Z

Test build #24799 has finished for PR 3721 at commit 46a2cd9.

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

AmplabJenkins · 2014-12-25T03:12:59Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24799/
Test PASSed.

tdas · 2014-12-25T03:50:14Z

Merged this, thanks very much!

…for-loop) in WriteAheadLogBasedBlockHandler Use `Future.zip` instead of `Future.flatMap`(for-loop). `zip` implies these two Futures will run concurrently, while `flatMap` usually means one Future depends on the other one. Author: zsxwing <zsxwing@gmail.com> Closes #3721 from zsxwing/SPARK-4873 and squashes the following commits: 46a2cd9 [zsxwing] Use Future.zip instead of Future.flatMap(for-loop) (cherry picked from commit b4d0db8) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

zsxwing changed the title ~~WriteAheadLogBasedBlockHandler improvement~~ [SPARK-4873][Streaming] WriteAheadLogBasedBlockHandler improvement Dec 17, 2014

zsxwing mentioned this pull request Dec 18, 2014

[SPARK-4790][STREAMING] Fix ReceivedBlockTrackerSuite waits for old file... #3726

Closed

harishreedharan reviewed Dec 18, 2014
View reviewed changes

JoshRosen mentioned this pull request Dec 18, 2014

[SPARK-3431] [WIP] Parallelize Scala/Java test execution #3564

Closed

tdas reviewed Dec 24, 2014
View reviewed changes

Use Future.zip instead of Future.flatMap(for-loop)

46a2cd9

zsxwing changed the title ~~[SPARK-4873][Streaming] WriteAheadLogBasedBlockHandler improvement~~ [SPARK-4873][Streaming] Use Future.zip instead of Future.flatMap(for-loop) in WriteAheadLogBasedBlockHandler Dec 25, 2014

asfgit closed this in b4d0db8 Dec 25, 2014

zsxwing deleted the SPARK-4873 branch December 25, 2014 03:53

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-4873][Streaming] Use `Future.zip` instead of `Future.flatMap`(for-loop) in WriteAheadLogBasedBlockHandler #3721

[SPARK-4873][Streaming] Use `Future.zip` instead of `Future.flatMap`(for-loop) in WriteAheadLogBasedBlockHandler #3721

zsxwing commented Dec 17, 2014

SparkQA commented Dec 17, 2014

SparkQA commented Dec 17, 2014

SparkQA commented Dec 17, 2014

AmplabJenkins commented Dec 17, 2014

SparkQA commented Dec 17, 2014

AmplabJenkins commented Dec 17, 2014

JoshRosen commented Dec 17, 2014

harishreedharan commented Dec 17, 2014

harishreedharan commented Dec 17, 2014

harishreedharan commented Dec 17, 2014

zsxwing commented Dec 18, 2014

zsxwing commented Dec 18, 2014

SparkQA commented Dec 18, 2014

SparkQA commented Dec 18, 2014

AmplabJenkins commented Dec 18, 2014

harishreedharan Dec 18, 2014

zsxwing Dec 18, 2014

harishreedharan Dec 18, 2014

zsxwing Dec 19, 2014

zsxwing Dec 19, 2014

tdas Dec 24, 2014

tdas commented Dec 24, 2014

zsxwing commented Dec 25, 2014

SparkQA commented Dec 25, 2014

tdas commented Dec 25, 2014

SparkQA commented Dec 25, 2014

AmplabJenkins commented Dec 25, 2014

tdas commented Dec 25, 2014

[SPARK-4873][Streaming] Use Future.zip instead of Future.flatMap(for-loop) in WriteAheadLogBasedBlockHandler #3721

[SPARK-4873][Streaming] Use Future.zip instead of Future.flatMap(for-loop) in WriteAheadLogBasedBlockHandler #3721

Conversation

zsxwing commented Dec 17, 2014

SparkQA commented Dec 17, 2014

SparkQA commented Dec 17, 2014

SparkQA commented Dec 17, 2014

AmplabJenkins commented Dec 17, 2014

SparkQA commented Dec 17, 2014

AmplabJenkins commented Dec 17, 2014

JoshRosen commented Dec 17, 2014

harishreedharan commented Dec 17, 2014

harishreedharan commented Dec 17, 2014

harishreedharan commented Dec 17, 2014

zsxwing commented Dec 18, 2014

zsxwing commented Dec 18, 2014

SparkQA commented Dec 18, 2014

SparkQA commented Dec 18, 2014

AmplabJenkins commented Dec 18, 2014

harishreedharan Dec 18, 2014

Choose a reason for hiding this comment

zsxwing Dec 18, 2014

Choose a reason for hiding this comment

harishreedharan Dec 18, 2014

Choose a reason for hiding this comment

zsxwing Dec 19, 2014

Choose a reason for hiding this comment

zsxwing Dec 19, 2014

Choose a reason for hiding this comment

tdas Dec 24, 2014

Choose a reason for hiding this comment

tdas commented Dec 24, 2014

zsxwing commented Dec 25, 2014

SparkQA commented Dec 25, 2014

tdas commented Dec 25, 2014

SparkQA commented Dec 25, 2014

AmplabJenkins commented Dec 25, 2014

tdas commented Dec 25, 2014

[SPARK-4873][Streaming] Use `Future.zip` instead of `Future.flatMap`(for-loop) in WriteAheadLogBasedBlockHandler #3721

[SPARK-4873][Streaming] Use `Future.zip` instead of `Future.flatMap`(for-loop) in WriteAheadLogBasedBlockHandler #3721