Merge pull request redpanda-data#29127 from bashtanov/coco-expose

Expose coordinated compaction details in debug endpoint

Author: bashtanov

src/v/cluster/cluster_utils.cc

@@ -16,6 +16,7 @@
 #include "cluster/snapshot.h"
 #include "cluster/types.h"
 #include "model/fips_config.h"
+#include "raft/consensus.h"
 #include "raft/consensus_utils.h"
 #include "raft/errc.h"
 #include "storage/disk_log_impl.h"
@@ -270,8 +271,15 @@ partition_state get_partition_state(ss::lw_shared_ptr<partition> partition) {
     state.revision_id = partition->get_revision_id();
     state.log_size_bytes = partition->size_bytes();
     state.non_log_disk_size_bytes = partition->non_log_disk_size_bytes();
+    auto& coco = partition->raft()->get_compaction_coordinator();
     state.max_tombstone_removable_offset
-      = partition->log()->stm_manager()->max_tombstone_remove_offset();
+      = coco.get_max_tombstone_remove_offset();
+    state.max_transaction_removable_offset
+      = coco.get_max_transaction_remove_offset();
+    state.max_cleanly_compacted_offset
+      = coco.get_local_max_cleanly_compacted_offset();
+    state.max_transaction_free_offset
+      = coco.get_local_max_transaction_free_offset();
     state.is_read_replica_mode_enabled
       = partition->is_read_replica_mode_enabled();
     state.is_remote_fetch_enabled = partition->is_remote_fetch_enabled();

src/v/cluster/types.h

@@ -2905,7 +2905,7 @@ struct partition_raft_state
 
 struct partition_state
   : serde::
-      envelope<partition_state, serde::version<2>, serde::compat_version<0>> {
+      envelope<partition_state, serde::version<3>, serde::compat_version<0>> {
     model::offset start_offset;
     model::offset committed_offset;
     model::offset last_stable_offset;
@@ -2924,6 +2924,9 @@ struct partition_state
     ss::sstring iceberg_mode;
     partition_raft_state raft_state;
     model::offset max_tombstone_removable_offset;
+    model::offset max_transaction_removable_offset;
+    model::offset max_cleanly_compacted_offset;
+    model::offset max_transaction_free_offset;
 
     auto serde_fields() {
         return std::tie(
@@ -2944,7 +2947,10 @@ struct partition_state
           read_replica_bucket,
           raft_state,
           iceberg_mode,
-          max_tombstone_removable_offset);
+          max_tombstone_removable_offset,
+          max_transaction_removable_offset,
+          max_cleanly_compacted_offset,
+          max_transaction_free_offset);
     }
 
     friend bool operator==(const partition_state&, const partition_state&)

src/v/redpanda/admin/api-doc/debug.json

@@ -1396,7 +1396,19 @@
                 },
                 "max_tombstone_removable_offset": {
                     "type": "long",
-                    "description": "Maximum offset up to which tombstones/transaction end markers can be removed."
+                    "description": "Maximum offset up to which tombstones can be removed (MTRO)."
+                },
+                "max_transaction_removable_offset": {
+                    "type": "long",
+                    "description": "Maximum offset up to which transaction end markers can be removed (MXRO)."
+                },
+                "max_cleanly_compacted_offset": {
+                    "type": "long",
+                    "description": "Maximum offset up to which this replica is cleanly compacted (MCCO)."
+                },
+                "max_transaction_free_offset": {
+                    "type": "long",
+                    "description": "Maximum offset up to which this replica is free of transaction fence batches and data (MXFO)."
                 },
                 "raft_state": {
                     "type": "raft_replica_state",

src/v/redpanda/admin/debug.cc

@@ -925,6 +925,11 @@ admin_server::get_partition_state_handler(
         replica.non_log_disk_size_bytes = state.non_log_disk_size_bytes;
         replica.max_tombstone_removable_offset
           = state.max_tombstone_removable_offset;
+        replica.max_transaction_removable_offset
+          = state.max_transaction_removable_offset;
+        replica.max_cleanly_compacted_offset
+          = state.max_cleanly_compacted_offset;
+        replica.max_transaction_free_offset = state.max_transaction_free_offset;
         replica.is_read_replica_mode_enabled
           = state.is_read_replica_mode_enabled;
         replica.read_replica_bucket = state.read_replica_bucket;

tests/rptest/tests/log_compaction_test.py

@@ -15,6 +15,7 @@
 
 from ducktape.mark import matrix
 from ducktape.utils.util import wait_until
+from ducktape.cluster.cluster import ClusterNode
 import ducktape.errors
 
 from rptest.clients.offline_log_viewer import OfflineLogViewer
@@ -951,6 +952,8 @@ def do_test_tx_control_batch_removal(self, test_case_name, test_case):
             f"Running test case {test_case_name} with topic {self.topic_spec.name}"
         )
 
+        self.check_all_offsets(lambda offset: offset <= 0)
+
         self.start_partition_movement()
         self.produce(test_case)
 
@@ -976,6 +979,31 @@ def produce_func():
 
         self.wait_for_all_tx_batches_removed(produce_func)
 
+        self.check_all_offsets(lambda offset: offset > 0)
+
+    def get_partition_state(self, node: ClusterNode | None = None):
+        state = self.redpanda._admin.get_partition_state(
+            namespace="kafka",
+            topic=self.topic_spec.name,
+            partition=0,
+            node=node,
+        )["replicas"]
+        self.redpanda.logger.debug(f"partition state: {state}")
+        return state
+
+    def check_all_offsets(self, predicate):
+        states = self.get_partition_state()
+        for state in states:
+            for offset_name in (
+                "max_tombstone_removable_offset",
+                "max_transaction_removable_offset",
+                "max_cleanly_compacted_offset",
+                "max_transaction_free_offset",
+            ):
+                if not predicate(state[offset_name]):
+                    return False
+        return True
+
 
 class LogCompactionTxRemovalTest(LogCompactionTxRemovalTestBase):
     def __init__(self, test_context):
@@ -1184,13 +1212,7 @@ def get_leader() -> Tuple[bool, int]:
 
         # make sure it sets MTRO beyond the segment with the commit batch
         def get_mtro():
-            state = self.redpanda._admin.get_partition_state(
-                namespace="kafka",
-                topic=self.topic_spec.name,
-                partition=0,
-                node=self.redpanda.get_node_by_id(leader_node),
-            )["replicas"]
-            self.redpanda.logger.debug(f"partition state: {state}")
+            state = self.get_partition_state(self.redpanda.get_node_by_id(leader_node))
             assert len(state) == 1
             return state[0]["max_tombstone_removable_offset"]