diff --git a/compaction/table_state.hh b/compaction/table_state.hh
index c11bd5a6e0..b74c25560d 100644
--- a/compaction/table_state.hh
+++ b/compaction/table_state.hh
@@ -50,6 +50,7 @@ public:
     virtual bool is_auto_compaction_disabled_by_user() const noexcept = 0;
     virtual const tombstone_gc_state& get_tombstone_gc_state() const noexcept = 0;
     virtual compaction_backlog_tracker& get_backlog_tracker() = 0;
+    virtual const std::string& get_group_id() const noexcept = 0;
 };
 
 }
diff --git a/replica/compaction_group.hh b/replica/compaction_group.hh
index 309327019b..fcaec7ca83 100644
--- a/replica/compaction_group.hh
+++ b/replica/compaction_group.hh
@@ -31,6 +31,7 @@ class compaction_group {
     table& _t;
     class table_state;
     std::unique_ptr<table_state> _table_state;
+    std::string _group_id;
     // Tokens included in this compaction_groups
     dht::token_range _token_range;
     compaction::compaction_strategy_state _compaction_strategy_state;
@@ -62,7 +63,11 @@ private:
 
     future<> delete_sstables_atomically(std::vector<sstables::shared_sstable> sstables_to_remove);
 public:
-    compaction_group(table& t, dht::token_range token_range);
+    compaction_group(table& t, std::string gid, dht::token_range token_range);
+
+    const std::string& get_group_id() const noexcept {
+        return _group_id;
+    }
 
     // Will stop ongoing compaction on behalf of this group, etc.
     future<> stop() noexcept;
diff --git a/replica/table.cc b/replica/table.cc
index 5ecc3c1a74..85a0bee1d8 100644
--- a/replica/table.cc
+++ b/replica/table.cc
@@ -526,8 +526,10 @@ std::vector<std::unique_ptr<compaction_group>> table::make_compaction_groups() {
     std::vector<std::unique_ptr<compaction_group>> ret;
     auto&& ranges = dht::split_token_range_msb(_x_log2_compaction_groups);
     tlogger.debug("Created {} compaction groups for {}.{}", ranges.size(), _schema->ks_name(), _schema->cf_name());
+    size_t i = 0;
     for (auto&& range : ranges) {
-        ret.emplace_back(std::make_unique<compaction_group>(*this, std::move(range)));
+        auto group_id = fmt::format("{}/{}", i++, ranges.size());
+        ret.emplace_back(std::make_unique<compaction_group>(*this, std::move(group_id), std::move(range)));
     }
     return ret;
 }
@@ -1487,9 +1489,10 @@ table::make_memtable_list(compaction_group& cg) {
     return make_lw_shared<memtable_list>(std::move(seal), std::move(get_schema), _config.dirty_memory_manager, _stats, _config.memory_compaction_scheduling_group);
 }
 
-compaction_group::compaction_group(table& t, dht::token_range token_range)
+compaction_group::compaction_group(table& t, std::string group_id, dht::token_range token_range)
     : _t(t)
     , _table_state(std::make_unique<table_state>(t, *this))
+    , _group_id(std::move(group_id))
     , _token_range(std::move(token_range))
     , _compaction_strategy_state(compaction::compaction_strategy_state::make(_t._compaction_strategy))
     , _memtables(_t._config.enable_disk_writes ? _t.make_memtable_list(*this) : _t.make_memory_only_memtable_list())
@@ -2798,6 +2801,9 @@ public:
     compaction_backlog_tracker& get_backlog_tracker() override {
         return _t._compaction_manager.get_backlog_tracker(*this);
     }
+    const std::string& get_group_id() const noexcept override {
+        return _cg.get_group_id();
+    }
 };
 
 compaction_backlog_tracker& compaction_group::get_backlog_tracker() {
diff --git a/test/lib/test_services.cc b/test/lib/test_services.cc
index 01210be55c..6f3c885318 100644
--- a/test/lib/test_services.cc
+++ b/test/lib/test_services.cc
@@ -48,6 +48,7 @@ class table_for_tests::table_state : public compaction::table_state {
     tombstone_gc_state _tombstone_gc_state;
     mutable compaction_backlog_tracker _backlog_tracker;
     compaction::compaction_strategy_state _compaction_strategy_state;
+    std::string _group_id;
 private:
     replica::table& table() const noexcept {
         return *_data.cf;
@@ -59,6 +60,7 @@ public:
             , _tombstone_gc_state(nullptr)
             , _backlog_tracker(get_compaction_strategy().make_backlog_tracker())
             , _compaction_strategy_state(compaction::compaction_strategy_state::make(get_compaction_strategy()))
+            , _group_id("table_for_tests::table_state")
     {
     }
     const schema_ptr& schema() const noexcept override {
@@ -116,6 +118,9 @@ public:
     compaction_backlog_tracker& get_backlog_tracker() override {
         return _backlog_tracker;
     }
+    const std::string& get_group_id() const noexcept override {
+        return _group_id;
+    }
 };
 
 table_for_tests::table_for_tests(sstables::sstables_manager& sstables_manager, schema_ptr s, std::optional<sstring> datadir)