bug: Prevent exec hang when task decode fails

[coderfender]

Which issue does this PR close?

Closes #1908

Rationale for this change

What changes are included in this PR?

Previously, all errors were masked in a blanket error but we now handle task serialization / gRPC errors in a separate path and send an event to fail the corresponding job without marking the executor as dead essentially failing the query. This should help with cluster stability and determinism with regards to task serialization errors

Are there any user-facing changes?

[andygrove]

Thanks @coderfender. The changes look reasonable, but is it possible to add tests to prevent regressions?

[coderfender]

Added tests

[phillipleblanc]

launch_tasks can combine stages from multiple jobs into one executor RPC, and then would now treat any InvalidArgument as a failure for every job in that RPC. The executor will process each job sequentially and then fail the first one with invalid argument:

ballista/executor/src/executor_server.rs:L850

   for multi_task in multi_tasks {
       let multi_task: Vec<TaskDefinition> = get_task_definition_vec(
           multi_task,
           // ...
       )
       .map_err(|e| Status::invalid_argument(format!("{e}")))?;

       for task in multi_task {
           task_sender
               .send(CuratorTaskDefinition {
                   scheduler_id: scheduler_id.clone(),
                   task,
               })
               .await
               .unwrap();
       }
   }

So this could cause tasks that were processed before the failing one to have already been enqueued, while skipping any tasks after the bad one. Then this code would mark every job as failed.

I think a better approach would be to isolate the failed jobs in launch_multi_task and return it as a separate list and only mark those specific ones as failed.

[phillipleblanc]

You could test this by having a test that sends a launch_tasks call with multiple jobs, where only one is invalid and verify that only that is marked as failed.

[coderfender]

Thank you for the comment @phillipleblanc . Let me try and isolate the test assertion to a single job in the unit test.