{"record":{"id":"4c3b6b308d14ea3b","repo":"risingwavelabs/risingwave","slug":"iceberg-compaction-task-is-already-running-for-sin","errorCode":null,"errorMessage":"iceberg compaction task is already running for sink {} (state=pending_dispatch, pending_commit_count_at_start={}, pending_commit_count={})","messagePattern":"iceberg compaction task is already running for sink (.+?) \\(state=pending_dispatch, pending_commit_count_at_start=(.+?), pending_commit_count=(.+?)\\)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"src/meta/src/manager/iceberg_compaction/schedule.rs","lineNumber":926,"sourceCode":"        if guard.manual_compaction_waiters.contains_key(&sink_id) {\n            return Err(anyhow!(\n                \"manual iceberg compaction is already waiting for sink {}\",\n                sink_id\n            )\n            .into());\n        }\n\n        if let Some(track) = guard.sink_schedules.get(&sink_id) {\n            if track.round_max_file_sequence_number.is_some() {\n                return Err(anyhow!(\n                    \"manual Full compaction is rejected while an automatic round is active for sink {}\",\n                    sink_id\n                )\n                .into());\n            }\n            match &track.state {\n                CompactionTrackState::PendingDispatch { attempt } => {\n                    return Err(anyhow!(\n                        \"iceberg compaction task is already running for sink {} \\\n                         (state=pending_dispatch, pending_commit_count_at_start={}, \\\n                         pending_commit_count={})\",\n                        sink_id,\n                        attempt.pending_commit_count_at_start,\n                        track.pending_commit_count\n                    )\n                    .into());\n                }\n                CompactionTrackState::InFlight {\n                    task_id,\n                    attempt,\n                    report_deadline,\n                    ..\n                } => {\n                    return Err(anyhow!(\n                        \"iceberg compaction task is already running for sink {} \\\n                         (state=in_flight, task_id={}, pending_commit_count_at_start={}, \\","sourceCodeStart":908,"sourceCodeEnd":944,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_compaction/schedule.rs#L908-L944","documentation":"The compaction scheduler rejects a new sink update because the sink's compaction track is in PendingDispatch state: a compaction task has been prepared but not yet dispatched to a worker, so another task cannot be started for the same sink. The error includes the pending commit counts captured at attempt start and now, for diagnosis.","triggerScenarios":"Calling start_manual_compaction (or otherwise applying a sink update) while guard.sink_schedules[sink].state is CompactionTrackState::PendingDispatch { attempt }.","commonSituations":"A compaction task was scheduled but its dispatch is delayed (worker contention, backpressure); operators issue a manual compaction that collides with the pending task. Commit count imbalance in the message hints at in-flight data commits the task is waiting on.","solutions":["Wait for the pending task to be dispatched and complete, then retry","Check meta logs for dispatch failures (worker unavailable) and ensure stream workers are healthy","If the pending state is stuck (no dispatch progress), restart the meta node to rebuild the in-memory schedule state","Investigate why pending_commit_count is not draining (downstream commit failures for the sink)"],"exampleFix":"// before\nlet rx = manager.start_manual_compaction(sink_id).await?; // Err: pending_dispatch\n// after\nmatch manager.start_manual_compaction(sink_id).await {\n    Ok(rx) => rx.await,\n    Err(e) if e.to_string().contains(\"already running\") => {\n        // back off and retry after current task completes\n        tokio::time::sleep(CHECK_INTERVAL).await;\n        retry(sink_id)\n    }\n    Err(e) => Err(e),\n}","handlingStrategy":"retry","validationCode":"// Retry only when the track is idle\nfn track_ready_for_update(track: Option<&CompactionTrack>) -> bool {\n    track.map(|t| matches!(t.state, CompactionTrackState::Idle { .. })).unwrap_or(true)\n}","typeGuard":null,"tryCatchPattern":"match start_manual_compaction(sink_id).await {\n    Err(e) if e.to_string().contains(\"already running\") && e.to_string().contains(\"pending_dispatch\") => {\n        tokio::time::sleep(RETRY_DELAY).await;\n        retry(sink_id)\n    }\n    other => other,\n}","preventionTips":["Avoid issuing manual compaction while a task is queued for dispatch","Monitor dispatch latency; long pending_dispatch suggests worker starvation","Fix worker availability so pending tasks dispatch quickly","Detect 'already running' in the message and back off exponentially"],"tags":["iceberg","compaction","already-running","concurrency"],"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"}