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

[zsxwing]

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]

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

• This patch merges cleanly.

[SparkQA]

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

• This patch merges cleanly.

[SparkQA]

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]

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]

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]

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]

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]

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]

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

[harishreedharan]

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

[zsxwing]

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]

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

[SparkQA]

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

• This patch merges cleanly.

[SparkQA]

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]

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]

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.

[zsxwing]

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

[harishreedharan]

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.

[zsxwing]

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.

[zsxwing]

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]

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

[tdas]

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]

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

[SparkQA]

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

• This patch merges cleanly.

[tdas]

LGTM. Will merge if tests pass.

[SparkQA]

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]

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]

Merged this, thanks very much!