risingwavelabs/risingwave · error · BatchError

task has been finished

Error message

task {:?} has been finished

What it means

Raised by BatchTaskExecution::check_if_aborted: when checking whether a task was aborted, the state lock shows TaskStatus::Finished instead, so the check bails with this message. It fires when a cancellation/abort request arrives for a task that already completed successfully before the abort could take effect. The offending input is a task_id whose abort is moot because the task finished first.

Solutions

1. Treat the error as benign in cancellation paths: the goal (no more work) is already achieved when the task finished
2. Check task status before issuing abort/cancel calls to skip already-finished tasks
3. Return the task's recorded result/output to the caller instead of the abort attempt's error
4. Suppress or downgrade this error in client-facing responses when the original request was a cancel

When it happens

Trigger: Thrown at src/batch/src/task/task_execution.rs:679 when the library encounters an invalid state.

Common situations: See trigger scenarios.

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/c8f0dd8a35504c71. Report an issue: GitHub.

Appendix: source

Thrown at src/batch/src/task/task_execution.rs:679

        let task_output = TaskOutput {
            receiver,
            output_id: output_id.try_into()?,
            failure: self.failure.clone(),
        };
        Ok(task_output)
    }

    pub fn check_if_running(&self) -> Result<()> {
        if *self.state.lock() != TaskStatus::Running {
            bail!("task {:?} is not running", self.get_task_id());
        }
        Ok(())
    }

    pub fn check_if_aborted(&self) -> Result<bool> {
        match *self.state.lock() {
            TaskStatus::Aborted => Ok(true),
            TaskStatus::Finished => bail!("task {:?} has been finished", self.get_task_id()),
            _ => Ok(false),
        }
    }

    /// Check the task status: whether has ended.
    pub fn is_end(&self) -> bool {
        let guard = self.state.lock();
        !(*guard == TaskStatus::Running || *guard == TaskStatus::Pending)
    }
}

impl BatchTaskExecution {
    pub(crate) fn set_heartbeat_join_handle(&self, join_handle: JoinHandle<()>) {
        *self.heartbeat_join_handle.lock() = Some(join_handle);
    }

    pub(crate) fn heartbeat_join_handle(&self) -> Option<JoinHandle<()>> {
        self.heartbeat_join_handle.lock().take()

View on GitHub (pinned to 6469eb736d)