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

[hunterhacker]

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 [bulk] When bulk stops a job itself, nothing anywhere records why #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.