{"record":{"id":"a8f5e9ee3e4633a2","repo":"risingwavelabs/risingwave","slug":"failed-to-cancel-streaming-job-id","errorCode":null,"errorMessage":"failed to cancel streaming job {id}","messagePattern":"failed to cancel streaming job (.+?)","errorType":"exception","errorClass":"MetaError","httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/stream_manager.rs","lineNumber":867,"sourceCode":"    /// 2. Send cancel message to recovered stream jobs (via `barrier_scheduler`).\n    ///\n    /// Cleanup of their state is handled by the caller after the drop command is collected.\n    pub async fn cancel_streaming_jobs(&self, job_ids: Vec<JobId>) -> MetaResult<Vec<JobId>> {\n        if job_ids.is_empty() {\n            return Ok(vec![]);\n        }\n\n        let _reschedule_job_lock = self.reschedule_lock_read_guard().await;\n        let (receivers, background_job_ids) = self.creating_job_info.cancel_jobs(job_ids).await?;\n\n        let futures = receivers.into_iter().map(|(id, receiver)| async move {\n            if let Ok(cancelled) = receiver.await\n                && cancelled\n            {\n                tracing::info!(\"canceled streaming job {id}\");\n                Ok(id)\n            } else {\n                Err(MetaError::from(anyhow::anyhow!(\n                    \"failed to cancel streaming job {id}\"\n                )))\n            }\n        });\n        let mut cancelled_ids = join_all(futures)\n            .await\n            .into_iter()\n            .collect::<MetaResult<Vec<_>>>()?;\n\n        // NOTE(kwannoel): For background_job_ids stream jobs that not tracked in streaming manager,\n        // we can directly cancel them by running the barrier command.\n        let futures = background_job_ids.into_iter().map(|id| async move {\n            let abort_result = self\n                .metadata_manager\n                .catalog_controller\n                .try_abort_creating_streaming_job(id, true)\n                .await?;\n            self.iceberg_compaction_manager","sourceCodeStart":849,"sourceCodeEnd":885,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/stream_manager.rs#L849-L885","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Retry the cancel/drop — a transient race with job completion is the most common cause.","Check the job's current state (SHOW JOBS / catalog): if it already finished or failed, cancellation is moot.","Inspect meta logs around the cancellation for dropped senders or barrier worker errors.","If the job is stuck, resolve the underlying barrier issue before retrying cancellation."],"exampleFix":"// before\nErr(MetaError::from(anyhow::anyhow!(\"failed to cancel streaming job {id}\")))\n// after (caller-side retry)\nfor _ in 0..3 {\n    match stream_manager.cancel_streaming_jobs(vec![id]).await {\n        Ok(_) => break,\n        Err(e) => { tokio::time::sleep(Duration::from_secs(2)).await; last = Some(e); }\n    }\n}","handlingStrategy":"retry","validationCode":"-- confirm the job is still active before cancelling\nSELECT job_id, state FROM rw_streaming_jobs WHERE job_id = <id> AND state NOT IN ('Created','Failed');","typeGuard":null,"tryCatchPattern":"for attempt in 0..3 {\n    match cancel(id).await {\n        Ok(_) => break,\n        Err(e) if e.to_string().contains(\"failed to cancel streaming job\") && attempt < 2 =>\n            tokio::time::sleep(Duration::from_secs(2)).await,\n        Err(e) => return Err(e),\n    }\n}","preventionTips":["Check job state before issuing CANCEL/DROP to avoid racing completion.","Retry cancellations idempotently; a false result may mean the job already ended.","Monitor meta-node failovers around maintenance windows when cancels are issued.","Log the job state at cancel time for easier triage."],"tags":["meta-node","cancellation","streaming-job","race-condition"],"backgroundTag":"invalid-state-transition","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}