rm_stm: put kafka offset cache behind feature manager

redpanda-data · Jul 8, 2022 · 46d2af3 · 46d2af3
1 parent 938f186
commit 46d2af3
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 3 deletions.
diff --git a/src/v/cluster/feature_table.cc b/src/v/cluster/feature_table.cc
@@ -30,6 +30,8 @@ std::string_view to_string_view(feature f) {
         return "serde_raft_0";
     case feature::license:
         return "license";
+    case feature::rm_stm_kafka_cache:
+        return "rm_stm_kafka_cache";
     case feature::test_alpha:
         return "__test_alpha";
     }
@@ -58,7 +60,7 @@ std::string_view to_string_view(feature_state::state s) {
 
 // The version that this redpanda node will report: increment this
 // on protocol changes to raft0 structures, like adding new services.
-static constexpr cluster_version latest_version = cluster_version{4};
+static constexpr cluster_version latest_version = cluster_version{5};
 
 feature_table::feature_table() {
     // Intentionally undocumented environment variable, only for use

diff --git a/src/v/cluster/feature_table.h b/src/v/cluster/feature_table.h
@@ -27,6 +27,7 @@ enum class feature : std::uint64_t {
     mtls_authentication = 0x8,
     serde_raft_0 = 0x10,
     license = 0x20,
+    rm_stm_kafka_cache = 0x40,
 
     // Dummy features for testing only
     test_alpha = uint64_t(1) << 63,
@@ -115,6 +116,12 @@ constexpr static std::array feature_schema{
     feature::license,
     feature_spec::available_policy::always,
     feature_spec::prepare_policy::always},
+  feature_spec{
+    cluster_version{5},
+    "rm_stm_kafka_cache",
+    feature::rm_stm_kafka_cache,
+    feature_spec::available_policy::always,
+    feature_spec::prepare_policy::always},
   feature_spec{
     cluster_version{2001},
     "__test_alpha",

diff --git a/src/v/cluster/rm_stm.cc b/src/v/cluster/rm_stm.cc
@@ -1960,7 +1960,12 @@ rm_stm::apply_snapshot(stm_snapshot_header hdr, iobuf&& tx_ss_buf) {
     _insync_offset = data.offset;
 }
 
-uint8_t rm_stm::active_snapshot_version() { return tx_snapshot_v1::version; }
+uint8_t rm_stm::active_snapshot_version() {
+    if (_feature_table.local().is_active(feature::rm_stm_kafka_cache)) {
+        return tx_snapshot::version;
+    }
+    return tx_snapshot_v1::version;
+}
 
 template<class T>
 void rm_stm::fill_snapshot_wo_seqs(T& snapshot) {

diff --git a/tests/rptest/tests/cluster_features_test.py b/tests/rptest/tests/cluster_features_test.py
@@ -42,7 +42,7 @@ def _assert_default_features(self):
         # This assertion will break each time we increment the value
         # of `latest_version` in the redpanda source.  Update it when
         # that happens.
-        assert features_response['cluster_version'] == 4
+        assert features_response['cluster_version'] == 5
 
         assert self._get_features_map(
             features_response)['central_config']['state'] == 'active'