Merge pull request #23097 from mmaslankaprv/CORE-6783-fix-decomm-test

c/members_table: include members table version in the snapshot

src/v/cluster/controller_snapshot.h

@@ -53,7 +53,7 @@ struct features_t
 
 struct members_t
   : public serde::
-      envelope<members_t, serde::version<1>, serde::compat_version<0>> {
+      envelope<members_t, serde::version<2>, serde::compat_version<0>> {
     struct node_t
       : serde::envelope<node_t, serde::version<0>, serde::compat_version<0>> {
         model::broker broker;
@@ -86,6 +86,9 @@ struct members_t
     absl::node_hash_map<model::node_id, update_t> in_progress_updates;
 
     model::offset first_node_operation_command_offset;
+    // revision of metadata table (offset of last applied cluster member
+    // command)
+    model::revision_id version{};
 
     friend bool operator==(const members_t&, const members_t&) = default;
 
@@ -97,7 +100,8 @@ struct members_t
           removed_nodes,
           removed_nodes_still_in_raft0,
           in_progress_updates,
-          first_node_operation_command_offset);
+          first_node_operation_command_offset,
+          version);
     }
 };
 

src/v/cluster/members_table.cc

@@ -293,14 +293,19 @@ void members_table::fill_snapshot(controller_snapshot& controller_snap) {
           controller_snapshot_parts::members_t::node_t{
             .broker = md.broker, .state = md.state});
     }
+    snap.version = _version;
 }
 
 void members_table::apply_snapshot(
   model::offset snap_offset, const controller_snapshot& controller_snap) {
-    _version = model::revision_id(snap_offset);
-
     const auto& snap = controller_snap.members;
 
+    // if version is present in snapshot use it, otherwise fallback to old
+    // behavior
+    _version = snap.version != model::revision_id{}
+                 ? snap.version
+                 : model::revision_id(snap_offset);
+
     // update the list of brokers
 
     cache_t old_nodes;