risingwavelabs/risingwave · warning

manual iceberg compaction is already waiting for sink {}

Error message

manual iceberg compaction is already waiting for sink {}

What it means

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.

Source

Thrown at src/meta/src/manager/iceberg_compaction/schedule.rs:909

    }

    pub(super) async fn start_manual_compaction(
        &self,
        sink_id: SinkId,
    ) -> MetaResult<oneshot::Receiver<MetaResult<IcebergCompactionTaskId>>> {
        let prepared_update = self
            .prepare_sink_update(
                sink_id,
                SinkUpdateKind::ManualForceCompaction {
                    task_type: TaskType::Full,
                },
                Instant::now(),
            )
            .await;
        let mut guard = self.inner.write();
        let now = Instant::now();
        if guard.manual_compaction_waiters.contains_key(&sink_id) {
            return Err(anyhow!(
                "manual iceberg compaction is already waiting for sink {}",
                sink_id
            )
            .into());
        }

        if let Some(track) = guard.sink_schedules.get(&sink_id) {
            if track.round_max_file_sequence_number.is_some() {
                return Err(anyhow!(
                    "manual Full compaction is rejected while an automatic round is active for sink {}",
                    sink_id
                )
                .into());
            }
            match &track.state {
                CompactionTrackState::PendingDispatch { attempt } => {
                    return Err(anyhow!(
                        "iceberg compaction task is already running for sink {} \

View on GitHub (pinned to 6469eb736d)

Solutions

1. Wait for the in-flight manual compaction to finish before re-triggering the same sink.
2. Check the task's status/result first; only re-trigger after completion or failure cleanup.
3. Serialize triggering at the application level (lock or dedupe by sink_id).
4. If no compaction is actually in flight, a leaked waiter remains; restart meta or report the cleanup bug.

Example fix

// before: concurrent double trigger
CALL rw_iceberg_compaction(42);
CALL rw_iceberg_compaction(42); -- already waiting
// after: await the first result, then retry
-- run once; poll task status before issuing again

// Application-level dedupe before triggering
if let Some(inflight) = compaction_inflight.get(&sink_id) {
    return Err(format!("compaction for sink {} already in flight", sink_id).into());
}
compaction_inflight.insert(sink_id);

match trigger_manual_compaction(sink_id).await {
    Err(e) if e.to_string().contains("already waiting") => {
        // poll the existing task's status instead of re-triggering
        poll_manual_compaction_status(sink_id).await
    }
    other => other,
}

When it happens

Trigger: 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.

Common situations: 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.

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

• Iceberg branch {} disappeared after manifest rewrite for sin
• Manual iceberg compaction task {} for sink {} failed: {}
• No iceberg compactor available
• Manual iceberg compaction waiter dropped unexpectedly for si
• Iceberg compaction task failed before dispatch for sink {}

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