group/tx/stm: limit local snapshot to max_removable_local_log_offset

The group_tx_tracker_stm snapshot could capture open transaction state
(begin_offsets/producer_states) at an offset where the corresponding
commit batch had not yet been written. If compaction later removed that
commit batch — which is allowed once max_removable_offset advances past
it — the open transaction could never be resolved on restart, permanently
blocking max_removable_offset and preventing further compaction.

The sequence:
1. Snapshot taken while a tx is open (fence at F, snapshot offset >= F)
2. Tx commits at offset C, max_removable advances past C
3. Compaction removes the commit batch at C
4. On restart, snapshot loads stale open tx at F, replay cannot find
   the commit -> max_removable stuck at prev(F) forever

Fix: snapshot at max_removable_local_log_offset with an empty
transactions map. Since this STM's sole purpose is tracking open
transactions for max_removable_local_log_offset, and closed transactions
leave no state, all meaningful state can be reconstructed from log
replay. Open transactions are re-discovered from fence batches in the
log, which are guaranteed to be present since compaction is bounded by
max_removable while the STM is live.

Also adds a regression test that reproduces the scenario by taking a
snapshot during an open tx, committing, compacting, re-persisting the
stale snapshot, and restarting.

To fix existing setups that have stale snapshots, this commit also bumps
supported_local_snapshot_version, this invalidates saved snapshots upon
upgrade and applies everything from log and reconstructs the correct
snapshots the next time with the newer logic.

(cherry picked from commit 45fff0d)

Author: bharathv

src/v/kafka/server/group_tx_tracker_stm.cc

@@ -161,10 +161,32 @@ group_tx_tracker_stm::apply_local_snapshot(
 
 ss::future<raft::stm_snapshot>
 group_tx_tracker_stm::take_local_snapshot(ssx::semaphore_units apply_units) {
-    // Copy over the snapshot state for a consistent view.
-    auto offset = last_applied_offset();
+    // Snapshot at max_removable_local_log_offset with open transaction state
+    // stripped. This prevents a bug where:
+    //
+    // 1. Snapshot captures an open tx (begin_offset in per_group_state)
+    // 2. The tx later commits, max_removable advances
+    // 3. Compaction removes the commit batch from the log
+    // 4. On restart, the stale open tx is loaded but its commit is gone
+    //    -> max_removable stuck permanently
+    //
+    // By definition, all open txs have begin_offset > max_removable, so they
+    // are all past the snapshot offset. On replay from offset+1, their fence
+    // batches are guaranteed to still be in the log (compaction is bounded by
+    // max_removable while the STM is live) and will re-establish the open tx
+    // state. Group existence (keys in _all_txs) must be preserved because
+    // maybe_add_tx_begin_offset() ignores fences for unknown groups, and the
+    // group_metadata batches that created them may be before the snapshot
+    // offset.
+    auto offset = max_removable_local_log_offset();
+
+    all_txs_t snap_txs;
+    for (const auto& [gid, _] : _all_txs) {
+        snap_txs[gid] = per_group_state{};
+    }
+
     snapshot snap{
-      .transactions{_all_txs},
+      .transactions{std::move(snap_txs)},
       .blocked_groups{
         std::from_range, _group_blocks | std::views::filter([](const auto& e) {
                              return e.second.is_blocked;

src/v/kafka/server/group_tx_tracker_stm.h

@@ -136,7 +136,7 @@ class group_tx_tracker_stm final
     }
 
 private:
-    static constexpr int8_t supported_local_snapshot_version = 1;
+    static constexpr int8_t supported_local_snapshot_version = 2;
     struct snapshot
       : serde::envelope<snapshot, serde::version<2>, serde::compat_version<0>> {
         all_txs_t transactions;