rm_stm: put kafka offset cache behind feature manager

(cherry picked from commit 8e7346d) Conflicts: src/v/cluster/feature_table.cc src/v/cluster/feature_table.h tests/rptest/tests/cluster_features_test.py
redpanda-data · Jul 13, 2022 · 74153f0 · 74153f0
1 parent 873762b
commit 74153f0
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
@@ -26,6 +26,8 @@ std::string_view to_string_view(feature f) {
         return "maintenance_mode";
     case feature::mtls_authentication:
         return "mtls_authentication";
+    case feature::rm_stm_kafka_cache:
+        return "rm_stm_kafka_cache";
     case feature::test_alpha:
         return "__test_alpha";
     }
@@ -34,7 +36,7 @@ std::string_view to_string_view(feature f) {
 
 // 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{3};
+static constexpr cluster_version latest_version = cluster_version{4};
 
 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
@@ -24,6 +24,7 @@ enum class feature : std::uint64_t {
     consumer_offsets = 0x2,
     maintenance_mode = 0x4,
     mtls_authentication = 0x8,
+    rm_stm_kafka_cache = 0x10,
 
     // Dummy features for testing only
     test_alpha = uint64_t(1) << 63,
@@ -100,6 +101,12 @@ constexpr static std::array feature_schema{
     feature::mtls_authentication,
     feature_spec::available_policy::explicit_only,
     feature_spec::prepare_policy::always},
+  feature_spec{
+    cluster_version{4},
+    "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
@@ -1959,7 +1959,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
@@ -40,7 +40,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'] == 3
+        assert features_response['cluster_version'] == 4
 
         assert self._get_features_map(
             features_response)['central_config']['state'] == 'active'