risingwavelabs/risingwave · error · MetaError

failed to cancel streaming job {id}

Error message

failed to cancel streaming job {id}

What it means

When cancelling streaming jobs, the meta node awaits each job's cancellation result over a channel. If the receiver yields an error (worker dropped/crashed) or reports cancelled == false, the cancellation did not complete and a MetaError with 'failed to cancel streaming job {id}' is returned for that id.

Source

Thrown at src/meta/src/stream/stream_manager.rs:867

    /// 2. Send cancel message to recovered stream jobs (via `barrier_scheduler`).
    ///
    /// Cleanup of their state is handled by the caller after the drop command is collected.
    pub async fn cancel_streaming_jobs(&self, job_ids: Vec<JobId>) -> MetaResult<Vec<JobId>> {
        if job_ids.is_empty() {
            return Ok(vec![]);
        }

        let _reschedule_job_lock = self.reschedule_lock_read_guard().await;
        let (receivers, background_job_ids) = self.creating_job_info.cancel_jobs(job_ids).await?;

        let futures = receivers.into_iter().map(|(id, receiver)| async move {
            if let Ok(cancelled) = receiver.await
                && cancelled
            {
                tracing::info!("canceled streaming job {id}");
                Ok(id)
            } else {
                Err(MetaError::from(anyhow::anyhow!(
                    "failed to cancel streaming job {id}"
                )))
            }
        });
        let mut cancelled_ids = join_all(futures)
            .await
            .into_iter()
            .collect::<MetaResult<Vec<_>>>()?;

        // NOTE(kwannoel): For background_job_ids stream jobs that not tracked in streaming manager,
        // we can directly cancel them by running the barrier command.
        let futures = background_job_ids.into_iter().map(|id| async move {
            let abort_result = self
                .metadata_manager
                .catalog_controller
                .try_abort_creating_streaming_job(id, true)
                .await?;
            self.iceberg_compaction_manager

View on GitHub (pinned to 6469eb736d)

Solutions

1. Retry the cancel/drop — a transient race with job completion is the most common cause.
2. Check the job's current state (SHOW JOBS / catalog): if it already finished or failed, cancellation is moot.
3. Inspect meta logs around the cancellation for dropped senders or barrier worker errors.
4. If the job is stuck, resolve the underlying barrier issue before retrying cancellation.

Example fix

// before
Err(MetaError::from(anyhow::anyhow!("failed to cancel streaming job {id}")))
// after (caller-side retry)
for _ in 0..3 {
    match stream_manager.cancel_streaming_jobs(vec![id]).await {
        Ok(_) => break,
        Err(e) => { tokio::time::sleep(Duration::from_secs(2)).await; last = Some(e); }
    }
}

-- confirm the job is still active before cancelling
SELECT job_id, state FROM rw_streaming_jobs WHERE job_id = <id> AND state NOT IN ('Created','Failed');

for attempt in 0..3 {
    match cancel(id).await {
        Ok(_) => break,
        Err(e) if e.to_string().contains("failed to cancel streaming job") && attempt < 2 =>
            tokio::time::sleep(Duration::from_secs(2)).await,
        Err(e) => return Err(e),
    }
}

When it happens

Trigger: Calling cancel_streaming_jobs for a job id where the cancel-future's receiver.await returns Err (sender side of the oneshot/channel dropped, e.g. barrier worker failed before replying) or replies cancelled=false (cancellation was not enacted before the job finished/failed).

Common situations: Cancelling a job that is concurrently finishing or already failed; meta node crash/failover losing the in-flight cancel; `CANCEL`/`DROP` issued right as a background job completes; recovering from a stuck barrier so the cancel never lands.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• database {database_id} does not exist while handling command
• Receive shutdown msg: {msg:?}
• empty meta addresses
• there should be only one load balance address
• failed to parse meta address `{1}`: {0}

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