awslabs / awslabs/amazon-dynamodb-tools

[bulk] The watchdog kills a run on the first OutOfMemoryError, even where Spark could have recovered

Open
#352 0 comments 0 reactions 0 assignees View on GitHub
bulk_executor help wanted
Dominant language
Python
Stars
181
Forks
50
Avg merge
4h 11m
Merged PRs (30d)
36

Description

## The policy today

`_watch_log_group` stops the Glue job the moment **any** stream logs a line matching
`OutOfMemoryError:` -- and it scans executor streams as well as the driver's. One executor
hitting a heap error therefore ends the whole run.

For a small job that is the right call. Measured on `--XNumberOfWorkers 2` (one executor,
four cores), from the executor's own log:

```
ERROR Executor:275 - Exception in task 0.0 in stage 2.0 (TID 200)
java.lang.OutOfMemoryError: Java heap space
ERROR Executor:267 - Executor launch task task 0.1 in stage 2.0 (TID 201) failed, reason:
Task ...TaskRunner@12270630 rejected from ...ThreadPoolExecutor@ca55f7c[Shutting down, ...]
ERROR Executor:267 - Executor launch task task 0.2 ... rejected from ...[Shutting down, ...]
ERROR Executor:267 - Executor launch task task 0.3 ... rejected from ...[Shutting down, ...]
```

All four attempts were consumed within the same second: the retries had nowhere to go
because the only executor was already shutting down. Stopping saved a couple of doomed
minutes.

## Where it may be wrong

The default is **220 workers**. An executor that dies there takes its tasks with it, but
Spark re-schedules them on the ~219 others, and `spark.task.maxFailures` allows four
attempts. A single skewed partition, or one executor unlucky enough to hold two large tasks
at once, can produce a heap error the run would have recovered from.

We kill it anyway -- including a `copy` or `fill` that is hours in. That trade (one OOM line
= abort) has never been measured at scale, and it is not obviously the right one.

## Worth measuring before changing

- Does a 220-worker run actually recover from a single executor OOM, or does the same
aggregation just OOM wherever it lands? (For a whole-table sort or `collect_list` the
answer is probably "dies everywhere", since the task is the same size on every executor.)
- Is a *driver* OOM ever recoverable? Almost certainly not -- worth keeping the immediate
stop for that one regardless.

## Possible shapes

- Distinguish driver from executor: stop immediately for the driver, and for an executor
only after N distinct occurrences or once Spark reports the stage aborted (`Task N in
stage M failed 4 times; aborting job` is logged, and is the point where recovery has
provably failed).
- Or keep the current behaviour and simply say so in the output -- "stopped after the first
executor ran out of memory; re-run with a larger --XWorkerType" -- which #345 now does.

Filed as a question, not a bug: the current behaviour is defensible, but it should be
deliberate rather than a side effect of matching one string in every stream.

Contributor guide

Open the contributing guide

Research direction

Start at `_watch_log_group`, which currently matches OutOfMemoryError lines across driver and executor streams. Measure the described 220-worker recovery case and compare it with driver OOM and Spark stage-abort messages; done means a documented policy decision supported by those observations, whether the current behavior is retained or changed.

Written by the indexing model from the issue text.

Assessment

Tech stack
aws, python, spark
Domain
distributed-systems, observability-sre
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Active
Clarity
Needs clarification
Newbie friendliness
30/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.