r/tests: added state machine manager test

Added state machine manager test waiting for batches to be applied after
each replicate. This test is designed to detect a situation in which
background apply fiber finishes but managed stm is still behind the
others.

Signed-off-by: Michal Maslanka <[email protected]>

Author: mmaslankaprv

src/v/raft/tests/stm_manager_test.cc

@@ -124,6 +124,31 @@ TEST_F_CORO(state_machine_fixture, test_apply_throwing_exception) {
         ASSERT_EQ_CORO(stm->state, expected);
     }
 }
+TEST_F_CORO(
+  state_machine_fixture, test_apply_throwing_exception_waiting_for_each_batch) {
+    /**
+     * Create 3 replicas group with simple_kv STM
+     */
+    create_nodes();
+    std::vector<ss::shared_ptr<simple_kv>> stms;
+
+    for (auto& [id, node] : nodes()) {
+        raft::state_machine_manager_builder builder;
+        auto kv_stm = builder.create_stm<simple_kv>(*node);
+        auto throwing_kv_stm = builder.create_stm<throwing_kv>(*node);
+        co_await node->init_and_start(all_vnodes(), std::move(builder));
+        stms.push_back(kv_stm);
+        stms.push_back(ss::dynamic_pointer_cast<simple_kv>(throwing_kv_stm));
+    }
+
+    auto expected = co_await build_random_state(5000, wait_for_each_batch::yes);
+
+    co_await wait_for_apply();
+
+    for (auto& stm : stms) {
+        ASSERT_EQ_CORO(stm->state, expected);
+    }
+}
 
 TEST_F_CORO(state_machine_fixture, test_recovery_without_snapshot) {
     /**

src/v/raft/tests/stm_test_fixture.h

@@ -34,6 +34,7 @@
 #include <seastar/core/shared_ptr.hh>
 #include <seastar/core/sstring.hh>
 #include <seastar/coroutine/parallel_for_each.hh>
+#include <seastar/util/bool_class.hh>
 #include <seastar/util/log.hh>
 
 #include <ostream>
@@ -142,6 +143,7 @@ struct simple_kv : public raft::state_machine_base {
     state_t state;
     raft_node_instance& raft_node;
 };
+using wait_for_each_batch = ss::bool_class<struct wait_for_each_tag>;
 
 struct state_machine_fixture : raft_fixture {
     ss::future<result<raft::replicate_result>>
@@ -172,7 +174,8 @@ struct state_machine_fixture : raft_fixture {
     }
 
     ss::future<absl::flat_hash_map<ss::sstring, value_entry>>
-    build_random_state(int op_cnt) {
+    build_random_state(
+      int op_cnt, wait_for_each_batch wait_for_each = wait_for_each_batch::no) {
         absl::flat_hash_map<ss::sstring, value_entry> state;
 
         for (int i = 0; i < op_cnt;) {
@@ -213,6 +216,10 @@ struct state_machine_fixture : raft_fixture {
               logger().debug,
               "replication result: [last_offset: {}]",
               result.value().last_offset);
+
+            if (wait_for_each) {
+                co_await wait_for_apply();
+            }
         }
         co_return state;
     }