risingwave_meta/manager/iceberg_compaction/
manual.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use anyhow::anyhow;
16use risingwave_connector::sink::catalog::SinkId;
17use risingwave_pb::iceberg_compaction::subscribe_iceberg_compaction_event_request::ReportTask as IcebergReportTask;
18use risingwave_pb::iceberg_compaction::subscribe_iceberg_compaction_event_request::report_task::Status as IcebergReportTaskStatus;
19use risingwave_pb::id::IcebergCompactionTaskId;
20
21use super::*;
22
23impl IcebergCompactionManager {
24    pub(super) fn complete_manual_task_waiter(
25        waiter: ManualCompactionWaiter,
26        report: &IcebergReportTask,
27    ) {
28        let status = IcebergReportTaskStatus::try_from(report.status)
29            .unwrap_or(IcebergReportTaskStatus::Unspecified);
30        let result = match status {
31            IcebergReportTaskStatus::Success => Ok(report.task_id),
32            IcebergReportTaskStatus::Drained
33            | IcebergReportTaskStatus::Busy
34            | IcebergReportTaskStatus::BlockedCapacity
35            | IcebergReportTaskStatus::BlockedStatistics
36            | IcebergReportTaskStatus::Failed
37            | IcebergReportTaskStatus::Unspecified => {
38                let message = report
39                    .error_message
40                    .clone()
41                    .unwrap_or_else(|| "manual iceberg compaction failed".to_owned());
42                Err(anyhow!(
43                    "Manual iceberg compaction task {} for sink {} failed: {}",
44                    report.task_id,
45                    report.sink_id,
46                    message
47                )
48                .into())
49            }
50        };
51
52        let _ = waiter.send(result);
53    }
54
55    pub async fn trigger_manual_compaction(
56        &self,
57        sink_id: SinkId,
58    ) -> MetaResult<IcebergCompactionTaskId> {
59        // Fast-fail before registering a waiter when no compactor can pull the
60        // scheduler task.
61        if self.iceberg_compactor_manager.compactor_num() == 0 {
62            return Err(anyhow!("No iceberg compactor available").into());
63        }
64
65        let waiter = self.start_manual_compaction(sink_id).await?;
66        // Drop this request's receiver before cleanup. A replacement request's
67        // live sender is never removed by an obsolete caller's cancellation.
68        let mut cleanup_guard = scopeguard::guard(Some(waiter), |waiter| {
69            drop(waiter);
70            self.cancel_closed_manual_compaction_waiter(sink_id);
71        });
72
73        tracing::info!(
74            "Manual compaction requested for sink {}, waiting for completion...",
75            sink_id
76        );
77
78        match cleanup_guard.as_mut().unwrap().await {
79            Ok(result) => {
80                let _ = scopeguard::ScopeGuard::into_inner(cleanup_guard);
81                result
82            }
83            Err(_) => {
84                let _ = scopeguard::ScopeGuard::into_inner(cleanup_guard);
85                Err(anyhow!(
86                    "Manual iceberg compaction waiter dropped unexpectedly for sink {}",
87                    sink_id,
88                )
89                .into())
90            }
91        }
92    }
93
94    pub(super) fn cancel_closed_manual_compaction_waiter(&self, sink_id: SinkId) {
95        let mut guard = self.inner.write();
96        if guard
97            .manual_compaction_waiters
98            .get(&sink_id)
99            .is_some_and(|sender| sender.is_closed())
100        {
101            guard.manual_compaction_waiters.remove(&sink_id);
102        }
103    }
104}
risingwave_meta/manager/iceberg_compaction/manual.rs

risingwave_meta/manager/iceberg_compaction/
manual.rs
