{"record":{"id":"8aac7f8a3045f1d2","repo":"risingwavelabs/risingwave","slug":"manual-iceberg-compaction-is-already-waiting-for-s","errorCode":null,"errorMessage":"manual iceberg compaction is already waiting for sink {}","messagePattern":"manual iceberg compaction is already waiting for sink (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"src/meta/src/manager/iceberg_compaction/schedule.rs","lineNumber":909,"sourceCode":"    }\n\n    pub(super) async fn start_manual_compaction(\n        &self,\n        sink_id: SinkId,\n    ) -> MetaResult<oneshot::Receiver<MetaResult<IcebergCompactionTaskId>>> {\n        let prepared_update = self\n            .prepare_sink_update(\n                sink_id,\n                SinkUpdateKind::ManualForceCompaction {\n                    task_type: TaskType::Full,\n                },\n                Instant::now(),\n            )\n            .await;\n        let mut guard = self.inner.write();\n        let now = Instant::now();\n        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 {} \\","sourceCodeStart":891,"sourceCodeEnd":927,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_compaction/schedule.rs#L891-L927","documentation":"start_manual_compaction registers one waiter per sink; if manual_compaction_waiters already contains the sink_id, a concurrent manual compaction is already in flight, and a second registration is rejected with this error to prevent waiter overwrite and duplicate tasks.","triggerScenarios":"Calling start_manual_compaction (via trigger_manual_compaction or the SQL CALL path) while another manual compaction for the same sink is still registered/waiting.","commonSituations":"User issues CALL rw_iceberg_compaction twice for the same sink before the first completes; concurrent dashboards/scripts both triggering compaction; a previous waiter not yet cleaned up after cancellation.","solutions":["Wait for the in-flight manual compaction to finish before re-triggering the same sink.","Check the task's status/result first; only re-trigger after completion or failure cleanup.","Serialize triggering at the application level (lock or dedupe by sink_id).","If no compaction is actually in flight, a leaked waiter remains; restart meta or report the cleanup bug."],"exampleFix":"// before: concurrent double trigger\nCALL rw_iceberg_compaction(42);\nCALL rw_iceberg_compaction(42); -- already waiting\n// after: await the first result, then retry\n-- run once; poll task status before issuing again","handlingStrategy":"validation","validationCode":"// Application-level dedupe before triggering\nif let Some(inflight) = compaction_inflight.get(&sink_id) {\n    return Err(format!(\"compaction for sink {} already in flight\", sink_id).into());\n}\ncompaction_inflight.insert(sink_id);","typeGuard":null,"tryCatchPattern":"match trigger_manual_compaction(sink_id).await {\n    Err(e) if e.to_string().contains(\"already waiting\") => {\n        // poll the existing task's status instead of re-triggering\n        poll_manual_compaction_status(sink_id).await\n    }\n    other => other,\n}","preventionTips":["Deduplicate manual compaction triggers per sink_id in tooling/scripts.","Poll task status instead of issuing repeated CALLs.","Confirm waiter cleanup on cancellation so leaked registrations do not block future runs."],"tags":["rust","meta-service","iceberg","compaction","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"}