[v26.1.x] cluster/rm_stm: preserve open transaction producers in local snapshots

src/v/cluster/rm_stm.cc

@@ -2134,7 +2134,9 @@ ss::future<raft::stm_snapshot> rm_stm::do_take_local_snapshot(
           snap.finished_requests, [start_kafka_offset](const auto& req) {
               return req.last_offset < start_kafka_offset;
           });
-        if (!snap.finished_requests.empty()) {
+        if (
+          !snap.finished_requests.empty()
+          || state->has_transaction_in_progress()) {
             stm_snapshot.producers.push_back(std::move(snap));
         }
     }

src/v/cluster/tests/rm_stm_test_fixture.h

@@ -100,6 +100,21 @@ struct rm_stm_test_fixture : simple_raft_fixture {
         return _stm->apply_local_snapshot(hdr, std::move(buf));
     }
 
+    // Tears down all infrastructure and recreates it from the same data
+    // directory. Invalidates all existing references to _stm and _raft.
+    void restart_stm_and_raft(
+      storage::ntp_config::default_overrides overrides = {}) {
+        stop_all();
+        producer_state_manager.stop().get();
+        producer_expiration_ms.stop().get();
+        max_concurent_producers.stop().get();
+        create_stm_and_start_raft(overrides);
+        _stm->testing_only_disable_auto_abort();
+        _stm->start().get();
+        wait_for_confirmed_leader();
+        wait_for_meta_initialized();
+    }
+
     auto wait_for_kafka_offset_apply(kafka::offset offset) {
         auto raft_offset = _stm->to_log_offset(offset);
         return _stm->wait(raft_offset, model::timeout_clock::now() + 10ms);

src/v/cluster/tests/rm_stm_tests.cc

@@ -1200,3 +1200,63 @@ FIXTURE_TEST(test_raft_snapshot_roundtrip, rm_stm_test_fixture) {
         BOOST_REQUIRE((bool)offset_r);
     }
 }
+
+// Ensures that a transaction whose begin marker is in a local snapshot, but
+// whose data batches aren't is still able to be committed and aborted after a
+// restart.
+FIXTURE_TEST(
+  test_local_snapshot_preserves_open_tx_producer, rm_stm_test_fixture) {
+    start_and_disable_auto_abort();
+    auto* stm = _stm.get();
+
+    auto pid1 = model::producer_identity{1, 0};
+    auto pid2 = model::producer_identity{2, 0};
+    auto tx_seq = model::tx_seq{0};
+
+    BOOST_REQUIRE(stm->begin_tx(pid1, tx_seq, timeout, model::partition_id(0))
+                    .get()
+                    .has_value());
+    BOOST_REQUIRE(stm->begin_tx(pid2, tx_seq, timeout, model::partition_id(0))
+                    .get()
+                    .has_value());
+
+    // Take a local snapshot after the fences but before data batches.
+    // Both producers have open transactions (status=initialized) but no
+    // finished_requests.
+    stm->write_local_snapshot().get();
+
+    // Replicate data batches for both producers.
+    auto r1 = replicate_all(*stm, make_batches(pid1, 0, 5, true)).get();
+    BOOST_REQUIRE(r1.has_value());
+    auto r2 = replicate_all(*stm, make_batches(pid2, 0, 5, true)).get();
+    BOOST_REQUIRE(r2.has_value());
+    auto last_data_offset = r2.value().last_offset;
+
+    // Restart the entire raft group. On start, the STM loads the local
+    // snapshot from disk and replays the log from the snapshot offset.
+    restart_stm_and_raft();
+    stm = _stm.get();
+
+    // Ensure the producers weren't lost from restarting.
+    BOOST_REQUIRE(producers().contains(pid1.get_id()));
+    BOOST_REQUIRE(producers().contains(pid2.get_id()));
+
+    // Ensure the LSO is held back by the open transactions.
+    auto lso = stm->last_stable_offset();
+    BOOST_REQUIRE_NE(lso, model::invalid_lso);
+    BOOST_REQUIRE_LE(lso, model::offset(last_data_offset()));
+
+    // Ensure pid1's transaction can be committed
+    auto commit_result = stm->commit_tx(pid1, tx_seq, 2'000ms).get();
+    BOOST_REQUIRE_EQUAL(commit_result, cluster::tx::errc::none);
+
+    // Ensure pid2's transaction can be aborted
+    auto abort_result
+      = stm->abort_tx(pid2, tx_seq, model::timeout_clock::duration{2'000ms})
+          .get();
+    BOOST_REQUIRE_EQUAL(abort_result, cluster::tx::errc::none);
+
+    RPTEST_REQUIRE_EVENTUALLY(10s, [stm, last_data_offset]() {
+        return model::offset(last_data_offset()) < stm->last_stable_offset();
+    });
+}