{"record":{"id":"f3739d8dfadc1751","repo":"risingwavelabs/risingwave","slug":"iceberg-compaction-task-is-already-running-for-sin-f3739d","errorCode":null,"errorMessage":"iceberg compaction task is already running for sink {} (state=in_flight, task_id={}, pending_commit_count_at_start={}, pending_commit_count={}, report_timeout_after_sec={})","messagePattern":"iceberg compaction task is already running for sink (.+?) \\(state=in_flight, task_id=(.+?), pending_commit_count_at_start=(.+?), pending_commit_count=(.+?), report_timeout_after_sec=(.+?)\\)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"src/meta/src/manager/iceberg_compaction/schedule.rs","lineNumber":942,"sourceCode":"            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={}, \\\n                         pending_commit_count={}, report_timeout_after_sec={})\",\n                        sink_id,\n                        task_id,\n                        attempt.pending_commit_count_at_start,\n                        track.pending_commit_count,\n                        report_deadline.saturating_duration_since(now).as_secs()\n                    )\n                    .into());\n                }\n                CompactionTrackState::Idle { .. } => {}\n            }\n        }\n\n        if self.apply_sink_update(&mut guard, prepared_update) {\n            let (tx, rx) = oneshot::channel();\n            guard.manual_compaction_waiters.insert(sink_id, tx);","sourceCodeStart":924,"sourceCodeEnd":960,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_compaction/schedule.rs#L924-L960","documentation":"The compaction scheduler rejects a request because the sink's compaction task is InFlight: a task with the given task_id has been dispatched to a worker and has not yet reported completion. The error reports task_id, pending commit counts, and the seconds remaining before the report deadline so callers know when the state may self-clean via timeout.","triggerScenarios":"Calling start_manual_compaction (or applying a sink update) while guard.sink_schedules[sink].state is CompactionTrackState::InFlight { task_id, attempt, report_deadline, .. }.","commonSituations":"A worker is still executing the compaction task; a manual compaction attempt collides with it. If report_timeout_after_sec is 0 or small, the task is about to be timed out and can be retried shortly; a consistently hung task points to worker failure.","solutions":["Wait until report_timeout_after_sec elapses or the task reports completion, then retry","Check meta logs and worker logs for task_id to see whether the worker is alive and progressing","If the worker died without reporting, the scheduler will time out the task and reset the track to Idle; retry after that","Investigate worker crash/network issues that prevent task reports"],"exampleFix":"// before\nlet rx = manager.start_manual_compaction(sink_id).await?; // Err: in_flight, timeout in Ns\n// after\n// Wait out the report deadline before retrying\ntokio::time::sleep(Duration::from_secs(timeout_after_sec + 1)).await;\nmanager.start_manual_compaction(sink_id).await?;","handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"match rx_or_err {\n    Err(e) if e.to_string().contains(\"state=in_flight\") => {\n        // parse report_timeout_after_sec from message and wait it out\n        let secs = parse_timeout_secs(&e.to_string());\n        tokio::time::sleep(Duration::from_secs(secs + 1)).await;\n        retry(sink_id)\n    }\n    other => other,\n}","preventionTips":["Wait for report_timeout_after_sec (given in the error) before retrying","Alert on tasks repeatedly hitting in_flight state without reporting","Ensure workers report task completion promptly; check worker-meta connectivity","Use the task_id in the error to look up the task's fate in logs"],"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"}