[SPARK-17485] Prevent failed remote reads of cached blocks from failing entire job #15037

JoshRosen · 2016-09-09T22:03:26Z

What changes were proposed in this pull request?

In Spark's RDD.getOrCompute we first try to read a local copy of a cached RDD block, then a remote copy, and only fall back to recomputing the block if no cached copy (local or remote) can be read. This logic works correctly in the case where no remote copies of the block exist, but if there are remote copies and reads of those copies fail (due to network issues or internal Spark bugs) then the BlockManager will throw a BlockFetchException that will fail the task (and which could possibly fail the whole job if the read failures keep occurring).

In the cases of TorrentBroadcast and task result fetching we really do want to fail the entire job in case no remote blocks can be fetched, but this logic is inappropriate for reads of cached RDD blocks because those can/should be recomputed in case cached blocks are unavailable.

Therefore, I think that the BlockManager.getRemoteBytes() method should never throw on remote fetch errors and, instead, should handle failures by returning None.

How was this patch tested?

Block manager changes should be covered by modified tests in BlockManagerSuite: the old tests expected exceptions to be thrown on failed remote reads, while the modified tests now expect None to be returned from the getRemote* method.

I also manually inspected all usages of BlockManager.getRemoteValues(), getRemoteBytes(), and get() to verify that they correctly pattern-match on the result and handle None. Note that these None branches are already exercised because the old getRemoteBytes returned None when no remote locations for the block could be found (which could occur if an executor died and its block manager de-registered with the master).

JoshRosen · 2016-09-09T22:05:43Z

core/src/main/scala/org/apache/spark/storage/BlockFetchException.scala

-import org.apache.spark.SparkException
-
-private[spark]
-case class BlockFetchException(messages: String, throwable: Throwable)


This exception was added in #7927. I think the problem was that #7927 was written with only TorrentBroadcast in mind, in which case it's fine to throw a fatal exception because the task is doomed to fail anyways since it can't fetch the broadcast variable and can't recompute it.

srinathshankar · 2016-09-09T22:54:59Z

How does the refreshing of locations here : https://github.com/apache/spark/pull/15037/files?diff=unified#diff-2b643ea78c1add0381754b1f47eec132R581
affect maxFetchFailures. Seems possible that if the number of new locations is smaller than the old, we repeatedly go over the same locations. Is that the intent ?

JoshRosen · 2016-09-09T23:09:22Z

Although the location refresh will reset runningFailureCount, the loop here should still terminate because it's based off of totalFailureCount, which is monotonically increasing. I agree that this is a bit confusing, though: it might make sense to automatically skip any locations which have been tried previously (even before the refresh), since the master might continue to report invalid locations.

JoshRosen · 2016-09-09T23:29:47Z

Ah, I see the issue: the problem is that we might end up repeatedly refreshing in case the number of returned locations is is much smaller than the max fetch attempt count.

I think that we might be able to greatly simplify things by removing a large portion of this retry / location refresh logic. Let me try to make some updates to see how feasible that will be.

JoshRosen · 2016-09-10T00:06:10Z

It looks like this location refresh logic was added in #11241 in order to resolve some issues which could occur on dynamic allocation clusters with high churn. While I definitely think that we should either fix the bug in this retry logic (or, better yet, move this logic higher up so that it's easier to understand and more obviously correct), I propose to do this in a separate patch so that this smaller fix can be backported independently into Spark 1.6.x. Let me go ahead and file a followup JIRA now.

SparkQA · 2016-09-10T00:15:12Z

Test build #65172 has finished for PR 15037 at commit 35efe6a.

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

SparkQA · 2016-09-10T01:28:12Z

Test build #65175 has finished for PR 15037 at commit d81e9c5.

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

SparkQA · 2016-09-10T02:25:51Z

Test build #65178 has finished for PR 15037 at commit 283ee79.

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

ericl · 2016-09-12T20:20:52Z

This LGTM

JoshRosen · 2016-09-12T22:43:39Z

Merging to master and branch-2.0.

…ng entire job ## What changes were proposed in this pull request? In Spark's `RDD.getOrCompute` we first try to read a local copy of a cached RDD block, then a remote copy, and only fall back to recomputing the block if no cached copy (local or remote) can be read. This logic works correctly in the case where no remote copies of the block exist, but if there _are_ remote copies and reads of those copies fail (due to network issues or internal Spark bugs) then the BlockManager will throw a `BlockFetchException` that will fail the task (and which could possibly fail the whole job if the read failures keep occurring). In the cases of TorrentBroadcast and task result fetching we really do want to fail the entire job in case no remote blocks can be fetched, but this logic is inappropriate for reads of cached RDD blocks because those can/should be recomputed in case cached blocks are unavailable. Therefore, I think that the `BlockManager.getRemoteBytes()` method should never throw on remote fetch errors and, instead, should handle failures by returning `None`. ## How was this patch tested? Block manager changes should be covered by modified tests in `BlockManagerSuite`: the old tests expected exceptions to be thrown on failed remote reads, while the modified tests now expect `None` to be returned from the `getRemote*` method. I also manually inspected all usages of `BlockManager.getRemoteValues()`, `getRemoteBytes()`, and `get()` to verify that they correctly pattern-match on the result and handle `None`. Note that these `None` branches are already exercised because the old `getRemoteBytes` returned `None` when no remote locations for the block could be found (which could occur if an executor died and its block manager de-registered with the master). Author: Josh Rosen <joshrosen@databricks.com> Closes #15037 from JoshRosen/SPARK-17485. (cherry picked from commit f9c580f) Signed-off-by: Josh Rosen <joshrosen@databricks.com>

…er put() exceptions ## What changes were proposed in this pull request? If a BlockManager `put()` call failed after the BlockManagerMaster was notified of a block's availability then incomplete cleanup logic in a `finally` block would never send a second block status method to inform the master of the block's unavailability. This, in turn, leads to fetch failures and used to be capable of causing complete job failures before #15037 was fixed. This patch addresses this issue via multiple small changes: - The `finally` block now calls `removeBlockInternal` when cleaning up from a failed `put()`; in addition to removing the `BlockInfo` entry (which was _all_ that the old cleanup logic did), this code (redundantly) tries to remove the block from the memory and disk stores (as an added layer of defense against bugs lower down in the stack) and optionally notifies the master of block removal (which now happens during exception-triggered cleanup). - When a BlockManager receives a request for a block that it does not have it will now notify the master to update its block locations. This ensures that bad metadata pointing to non-existent blocks will eventually be fixed. Note that I could have implemented this logic in the block manager client (rather than in the remote server), but that would introduce the problem of distinguishing between transient and permanent failures; on the server, however, we know definitively that the block isn't present. - Catch `NonFatal` instead of `Exception` to avoid swallowing `InterruptedException`s thrown from synchronous block replication calls. This patch depends upon the refactorings in #15036, so that other patch will also have to be backported when backporting this fix. For more background on this issue, including example logs from a real production failure, see [SPARK-17484](https://issues.apache.org/jira/browse/SPARK-17484). ## How was this patch tested? Two new regression tests in BlockManagerSuite. Author: Josh Rosen <joshrosen@databricks.com> Closes #15085 from JoshRosen/SPARK-17484. (cherry picked from commit 1202075) Signed-off-by: Josh Rosen <joshrosen@databricks.com>

…er put() exceptions ## What changes were proposed in this pull request? If a BlockManager `put()` call failed after the BlockManagerMaster was notified of a block's availability then incomplete cleanup logic in a `finally` block would never send a second block status method to inform the master of the block's unavailability. This, in turn, leads to fetch failures and used to be capable of causing complete job failures before #15037 was fixed. This patch addresses this issue via multiple small changes: - The `finally` block now calls `removeBlockInternal` when cleaning up from a failed `put()`; in addition to removing the `BlockInfo` entry (which was _all_ that the old cleanup logic did), this code (redundantly) tries to remove the block from the memory and disk stores (as an added layer of defense against bugs lower down in the stack) and optionally notifies the master of block removal (which now happens during exception-triggered cleanup). - When a BlockManager receives a request for a block that it does not have it will now notify the master to update its block locations. This ensures that bad metadata pointing to non-existent blocks will eventually be fixed. Note that I could have implemented this logic in the block manager client (rather than in the remote server), but that would introduce the problem of distinguishing between transient and permanent failures; on the server, however, we know definitively that the block isn't present. - Catch `NonFatal` instead of `Exception` to avoid swallowing `InterruptedException`s thrown from synchronous block replication calls. This patch depends upon the refactorings in #15036, so that other patch will also have to be backported when backporting this fix. For more background on this issue, including example logs from a real production failure, see [SPARK-17484](https://issues.apache.org/jira/browse/SPARK-17484). ## How was this patch tested? Two new regression tests in BlockManagerSuite. Author: Josh Rosen <joshrosen@databricks.com> Closes #15085 from JoshRosen/SPARK-17484.

…ng entire job ## What changes were proposed in this pull request? In Spark's `RDD.getOrCompute` we first try to read a local copy of a cached RDD block, then a remote copy, and only fall back to recomputing the block if no cached copy (local or remote) can be read. This logic works correctly in the case where no remote copies of the block exist, but if there _are_ remote copies and reads of those copies fail (due to network issues or internal Spark bugs) then the BlockManager will throw a `BlockFetchException` that will fail the task (and which could possibly fail the whole job if the read failures keep occurring). In the cases of TorrentBroadcast and task result fetching we really do want to fail the entire job in case no remote blocks can be fetched, but this logic is inappropriate for reads of cached RDD blocks because those can/should be recomputed in case cached blocks are unavailable. Therefore, I think that the `BlockManager.getRemoteBytes()` method should never throw on remote fetch errors and, instead, should handle failures by returning `None`. ## How was this patch tested? Block manager changes should be covered by modified tests in `BlockManagerSuite`: the old tests expected exceptions to be thrown on failed remote reads, while the modified tests now expect `None` to be returned from the `getRemote*` method. I also manually inspected all usages of `BlockManager.getRemoteValues()`, `getRemoteBytes()`, and `get()` to verify that they correctly pattern-match on the result and handle `None`. Note that these `None` branches are already exercised because the old `getRemoteBytes` returned `None` when no remote locations for the block could be found (which could occur if an executor died and its block manager de-registered with the master). Author: Josh Rosen <joshrosen@databricks.com> Closes apache#15037 from JoshRosen/SPARK-17485.

…er put() exceptions ## What changes were proposed in this pull request? If a BlockManager `put()` call failed after the BlockManagerMaster was notified of a block's availability then incomplete cleanup logic in a `finally` block would never send a second block status method to inform the master of the block's unavailability. This, in turn, leads to fetch failures and used to be capable of causing complete job failures before apache#15037 was fixed. This patch addresses this issue via multiple small changes: - The `finally` block now calls `removeBlockInternal` when cleaning up from a failed `put()`; in addition to removing the `BlockInfo` entry (which was _all_ that the old cleanup logic did), this code (redundantly) tries to remove the block from the memory and disk stores (as an added layer of defense against bugs lower down in the stack) and optionally notifies the master of block removal (which now happens during exception-triggered cleanup). - When a BlockManager receives a request for a block that it does not have it will now notify the master to update its block locations. This ensures that bad metadata pointing to non-existent blocks will eventually be fixed. Note that I could have implemented this logic in the block manager client (rather than in the remote server), but that would introduce the problem of distinguishing between transient and permanent failures; on the server, however, we know definitively that the block isn't present. - Catch `NonFatal` instead of `Exception` to avoid swallowing `InterruptedException`s thrown from synchronous block replication calls. This patch depends upon the refactorings in apache#15036, so that other patch will also have to be backported when backporting this fix. For more background on this issue, including example logs from a real production failure, see [SPARK-17484](https://issues.apache.org/jira/browse/SPARK-17484). ## How was this patch tested? Two new regression tests in BlockManagerSuite. Author: Josh Rosen <joshrosen@databricks.com> Closes apache#15085 from JoshRosen/SPARK-17484.

…ng entire job In Spark's `RDD.getOrCompute` we first try to read a local copy of a cached RDD block, then a remote copy, and only fall back to recomputing the block if no cached copy (local or remote) can be read. This logic works correctly in the case where no remote copies of the block exist, but if there _are_ remote copies and reads of those copies fail (due to network issues or internal Spark bugs) then the BlockManager will throw a `BlockFetchException` that will fail the task (and which could possibly fail the whole job if the read failures keep occurring). In the cases of TorrentBroadcast and task result fetching we really do want to fail the entire job in case no remote blocks can be fetched, but this logic is inappropriate for reads of cached RDD blocks because those can/should be recomputed in case cached blocks are unavailable. Therefore, I think that the `BlockManager.getRemoteBytes()` method should never throw on remote fetch errors and, instead, should handle failures by returning `None`. Block manager changes should be covered by modified tests in `BlockManagerSuite`: the old tests expected exceptions to be thrown on failed remote reads, while the modified tests now expect `None` to be returned from the `getRemote*` method. I also manually inspected all usages of `BlockManager.getRemoteValues()`, `getRemoteBytes()`, and `get()` to verify that they correctly pattern-match on the result and handle `None`. Note that these `None` branches are already exercised because the old `getRemoteBytes` returned `None` when no remote locations for the block could be found (which could occur if an executor died and its block manager de-registered with the master). Author: Josh Rosen <joshrosen@databricks.com> Closes apache#15037 from JoshRosen/SPARK-17485.

…ng entire job (branch-1.6 backport) This patch is a branch-1.6 backport of #15037: ## What changes were proposed in this pull request? In Spark's `RDD.getOrCompute` we first try to read a local copy of a cached RDD block, then a remote copy, and only fall back to recomputing the block if no cached copy (local or remote) can be read. This logic works correctly in the case where no remote copies of the block exist, but if there _are_ remote copies and reads of those copies fail (due to network issues or internal Spark bugs) then the BlockManager will throw a `BlockFetchException` that will fail the task (and which could possibly fail the whole job if the read failures keep occurring). In the cases of TorrentBroadcast and task result fetching we really do want to fail the entire job in case no remote blocks can be fetched, but this logic is inappropriate for reads of cached RDD blocks because those can/should be recomputed in case cached blocks are unavailable. Therefore, I think that the `BlockManager.getRemoteBytes()` method should never throw on remote fetch errors and, instead, should handle failures by returning `None`. ## How was this patch tested? Block manager changes should be covered by modified tests in `BlockManagerSuite`: the old tests expected exceptions to be thrown on failed remote reads, while the modified tests now expect `None` to be returned from the `getRemote*` method. I also manually inspected all usages of `BlockManager.getRemoteValues()`, `getRemoteBytes()`, and `get()` to verify that they correctly pattern-match on the result and handle `None`. Note that these `None` branches are already exercised because the old `getRemoteBytes` returned `None` when no remote locations for the block could be found (which could occur if an executor died and its block manager de-registered with the master). Author: Josh Rosen <joshrosen@databricks.com> Closes #15186 from JoshRosen/SPARK-17485-branch-1.6-backport.

…ng entire job (branch-1.6 backport) This patch is a branch-1.6 backport of apache#15037: ## What changes were proposed in this pull request? In Spark's `RDD.getOrCompute` we first try to read a local copy of a cached RDD block, then a remote copy, and only fall back to recomputing the block if no cached copy (local or remote) can be read. This logic works correctly in the case where no remote copies of the block exist, but if there _are_ remote copies and reads of those copies fail (due to network issues or internal Spark bugs) then the BlockManager will throw a `BlockFetchException` that will fail the task (and which could possibly fail the whole job if the read failures keep occurring). In the cases of TorrentBroadcast and task result fetching we really do want to fail the entire job in case no remote blocks can be fetched, but this logic is inappropriate for reads of cached RDD blocks because those can/should be recomputed in case cached blocks are unavailable. Therefore, I think that the `BlockManager.getRemoteBytes()` method should never throw on remote fetch errors and, instead, should handle failures by returning `None`. ## How was this patch tested? Block manager changes should be covered by modified tests in `BlockManagerSuite`: the old tests expected exceptions to be thrown on failed remote reads, while the modified tests now expect `None` to be returned from the `getRemote*` method. I also manually inspected all usages of `BlockManager.getRemoteValues()`, `getRemoteBytes()`, and `get()` to verify that they correctly pattern-match on the result and handle `None`. Note that these `None` branches are already exercised because the old `getRemoteBytes` returned `None` when no remote locations for the block could be found (which could occur if an executor died and its block manager de-registered with the master). Author: Josh Rosen <joshrosen@databricks.com> Closes apache#15186 from JoshRosen/SPARK-17485-branch-1.6-backport. (cherry picked from commit 94524ce)

Remove throw which caused the bug.

35efe6a

JoshRosen reviewed Sep 9, 2016
View reviewed changes

logError -> logWarning

d81e9c5

Update BlockManager.scala

283ee79

asfgit closed this in f9c580f Sep 12, 2016

JoshRosen deleted the SPARK-17485 branch September 13, 2016 20:00

JoshRosen mentioned this pull request Sep 13, 2016

[SPARK-17484] Prevent invalid block locations from being reported after put() exceptions #15085

Closed

JoshRosen mentioned this pull request Sep 21, 2016

[SPARK-17485] Prevent failed remote reads of cached blocks from failing entire job (branch-1.6 backport) #15186

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[SPARK-17485] Prevent failed remote reads of cached blocks from failing entire job #15037

[SPARK-17485] Prevent failed remote reads of cached blocks from failing entire job #15037

Uh oh!

JoshRosen commented Sep 9, 2016

Uh oh!

JoshRosen Sep 9, 2016

Uh oh!

srinathshankar commented Sep 9, 2016

Uh oh!

JoshRosen commented Sep 9, 2016

Uh oh!

JoshRosen commented Sep 9, 2016

Uh oh!

JoshRosen commented Sep 10, 2016

Uh oh!

SparkQA commented Sep 10, 2016

Uh oh!

SparkQA commented Sep 10, 2016

Uh oh!

SparkQA commented Sep 10, 2016

Uh oh!

ericl commented Sep 12, 2016

Uh oh!

JoshRosen commented Sep 12, 2016

Uh oh!

Uh oh!

[SPARK-17485] Prevent failed remote reads of cached blocks from failing entire job #15037

[SPARK-17485] Prevent failed remote reads of cached blocks from failing entire job #15037

Uh oh!

Conversation

JoshRosen commented Sep 9, 2016

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

JoshRosen Sep 9, 2016

Choose a reason for hiding this comment

Uh oh!

srinathshankar commented Sep 9, 2016

Uh oh!

JoshRosen commented Sep 9, 2016

Uh oh!

JoshRosen commented Sep 9, 2016

Uh oh!

JoshRosen commented Sep 10, 2016

Uh oh!

SparkQA commented Sep 10, 2016

Uh oh!

SparkQA commented Sep 10, 2016

Uh oh!

SparkQA commented Sep 10, 2016

Uh oh!

ericl commented Sep 12, 2016

Uh oh!

JoshRosen commented Sep 12, 2016

Uh oh!

Uh oh!