[SPARK-2670] FetchFailedException should be thrown when local fetch has failed

[sarutak]

No description provided.

[pwendell]

Thanks - this is a good idea. Two questions: (a) what type of exception have you seen here? (b) could you add a unit test for this? Jenkins, test this please.

[SparkQA]

QA tests have started for PR 1578. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17166/consoleFull

[SparkQA]

QA results for PR 1578:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17166/consoleFull

[mateiz]

Small code style thing, add a space before the {

[sarutak]

@pwendell I found this issue when I simulated disk fault. When shuffle____ cannot be open successfully, FileNotFoundException was thrown from the constructor of RandomAccessFile in DiskStore#getBytes.

Yes, I will add test cases later.

[witgo]

Here also should throw an FetchFailedException?

    override def next(): (BlockId, Option[Iterator[Any]]) = {
      resultsGotten += 1
      val startFetchWait = System.currentTimeMillis()
      val result = results.take()
      val stopFetchWait = System.currentTimeMillis()
      _fetchWaitTime += (stopFetchWait - startFetchWait)
      if (!result.failed) bytesInFlight -= result.size
      while (!fetchRequests.isEmpty &&
        (bytesInFlight == 0 || bytesInFlight + fetchRequests.front.size <= maxBytesInFlight)) {
        sendRequest(fetchRequests.dequeue())
      }
      (result.blockId, if (result.failed) None else Some(result.deserialize()))
    }

[sarutak]

I've modified BasicBlockFetcherIterator to fail fast and added test cases.

[mateiz]

I wouldn't do drop and such on a ConcurrentQueue, since it might drop stuff other threads were adding. Just do a results.put on the failed block and don't worry about dropping other ones. You can actually move the try/catch into the for loop and add a "return" at the bottom of the catch after adding this failing FetchResult.

[sarutak]

Thank you for your comment, @mateiz .

I wouldn't do drop and such on a ConcurrentQueue, since it might drop stuff other threads were adding. Just do a results.put on the failed block and don't worry about dropping other ones. You can actually move the try/catch into the for loop and add a "return" at the bottom of the catch after adding this failing FetchResult.

But, if it returns from getLocalBlocks immediately rest of FetchResults is not set to results, and we waits on results.take() in next method forever right? results is a instance of LinkedBlockingQueue and take method is blocking method.

[mateiz]

I thought next() would return a failure block, and then the caller of BlockFetcherIterator will just stop. Did you see it not doing that? I think all you have to do is put one FetchResult with size = -1 in the queue and return, and everything will be fine.

[sarutak]

I thought wrong. Exactly, in current usage of BlockFetcherIterator, next() is not invoked after FetchFailedException has been thrown.
I wonder it's a little bit problem that we can invoke next() after FetchFailedException even if there are no such usages in current implementation.
I think it's better to prohibit invoking next() after FetchFailedException to clearly express the correct usage of the method.

[mateiz]

Thanks for adding the test! I had one more comment on using drop() on the concurrent queue -- it seems like it might be troublesome. I'd rather just put the failed result and exit from getLocalBlocks

[sarutak]

localBlocksToFetch is a instance of ArrayBuffer, not concurrent queue and it is used from only one thread right?

[mateiz]

Jenkins, retest this please

[SparkQA]

QA tests have started for PR 1578. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17618/consoleFull

[SparkQA]

QA results for PR 1578:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17618/consoleFull

[mateiz]

Thanks for the changes! I've merged this in.