Re: [PR] bug: Prevent exec hang when task decode fails [datafusion-ballista]

["milenkovicm (via GitHub)" <gi...@apache.org>]

milenkovicm commented on code in PR #2016:
URL: https://github.com/apache/datafusion-ballista/pull/2016#discussion_r3705763168


##########
ballista/scheduler/src/state/task_manager.rs:
##########
@@ -106,10 +106,10 @@ impl TaskLauncher for DefaultTaskLauncher {
                 executor.id, tasks_ids
             );
         }
-        executor_manager
+        let res = executor_manager
             .launch_multi_task(&executor.id, tasks, self.scheduler_id.clone())
             .await?;
-        Ok(())
+        Ok(res)

Review Comment:
   same comment here, its a bit confusing that there Ok result with failing jobs, no documentation explains return type 



##########
ballista/scheduler/src/test_utils.rs:
##########
@@ -377,7 +377,34 @@ impl TaskLauncher for VirtualTaskLauncher {
             .await
             .map_err(|e| {
                 BallistaError::Internal(format!("Error sending task status: {e:?}"))
-            })
+            })?;
+        Ok(vec![])
+    }
+}
+
+/// Launcher that reports every job in the batch as rejected via the
+/// `failed_jobs` channel, simulating an executor that cannot decode/validate
+/// the task (see issue #1908). The RPC itself succeeds; the jobs are failed
+/// individually rather than the whole batch.
+#[derive(Default)]
+pub struct RejectingTaskLauncher {}
+
+#[async_trait::async_trait]
+impl TaskLauncher for RejectingTaskLauncher {
+    async fn launch_tasks(
+        &self,
+        _executor: &ExecutorMetadata,
+        tasks: Vec<MultiTaskDefinition>,
+        _executor_manager: &ExecutorManager,
+    ) -> Result<Vec<JobId>> {

Review Comment:
   should we return set if we do not expect duplicates  



##########
ballista/core/proto/ballista.proto:
##########
@@ -980,6 +980,7 @@ message LaunchTaskResult {
 
 message LaunchMultiTaskResult {
   bool success = 1;
+  repeated string failed_jobs = 2;
   // TODO when part of the task set are scheduled successfully

Review Comment:
   i guess this comment is irrelevant now, could we remove it ?



##########
ballista/scheduler/src/state/executor_manager.rs:
##########
@@ -408,23 +408,19 @@ impl ExecutorManager {
         executor_id: &str,
         multi_tasks: Vec<MultiTaskDefinition>,
         scheduler_id: String,
-    ) -> Result<()> {
+    ) -> Result<Vec<JobId>> {

Review Comment:
   this result is a bit confusing, we're return ok with list of failed jobs. Do we need to return error in this case, or some other indicator that some of the jobs failed 



##########
ballista/executor/src/executor_server.rs:
##########
@@ -914,7 +923,10 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> ExecutorGrpc
                     .unwrap();
             }
         }
-        Ok(Response::new(LaunchMultiTaskResult { success: true }))
+        Ok(Response::new(LaunchMultiTaskResult {
+            success: true,

Review Comment:
   its a bit confusing to have success: true but have a failing jobs, do we need success property ? 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org