fix(compaction): avoid merge a emergency group

src/meta/src/hummock/manager/compaction/compaction_group_schedule.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
-use std::ops::DerefMut;
+use std::ops::{Deref, DerefMut};
 use std::sync::Arc;
 
 use bytes::Bytes;
@@ -33,7 +33,7 @@ use risingwave_pb::hummock::{
 };
 use thiserror_ext::AsReport;
 
-use super::CompactionGroupStatistic;
+use super::{check_cg_write_limit, CompactionGroupStatistic};
 use crate::hummock::error::{Error, Result};
 use crate::hummock::manager::transaction::HummockVersionTransaction;
 use crate::hummock::manager::{commit_multi_var, HummockManager};
@@ -929,6 +929,41 @@ impl HummockManager {
             )));
         }
 
+        {
+            // TODO: check emergency state
+            let versioning_guard = self.versioning.read().await;
+            if check_cg_write_limit(
+                versioning_guard
+                    .current_version
+                    .get_compaction_group_levels(group.group_id),
+                group.compaction_group_config.compaction_config().deref(),
+            )
+            .is_write_stop()
+            {
+                return Err(Error::CompactionGroup(format!(
+                    "Not Merge write limit group {} next group {}",
+                    group.group_id, next_group.group_id
+                )));
+            }
+
+            if check_cg_write_limit(
+                versioning_guard
+                    .current_version
+                    .get_compaction_group_levels(next_group.group_id),
+                next_group
+                    .compaction_group_config
+                    .compaction_config()
+                    .deref(),
+            )
+            .is_write_stop()
+            {
+                return Err(Error::CompactionGroup(format!(
+                    "Not Merge write limit next group {} group {}",
+                    next_group.group_id, group.group_id
+                )));
+            }
+        }
+
         match self
             .merge_compaction_group(group.group_id, next_group.group_id)
             .await