Merge pull request #29609 from WillemKauf/tx_compact_fix

[CORE-8908] `cluster`: don't compact away batches in the idempotency window in `rm_stm`

Author: WillemKauf

src/v/cluster/producer_state.cc

@@ -235,6 +235,14 @@ void requests::stm_apply(
     }
 }
 
+bool requests::has_request_for_seq_range(seq_t first, seq_t last) const {
+    auto matches = [first, last](const request_ptr& req) {
+        return req->first_sequence() == first && req->last_sequence() == last;
+    };
+    return std::ranges::any_of(_inflight_requests, matches)
+           || std::ranges::any_of(_finished_requests, matches);
+}
+
 void requests::gc_requests_from_older_terms(model::term_id current_term) {
     while (!_inflight_requests.empty()
            && _inflight_requests.front()->_term < current_term) {

src/v/cluster/producer_state.h

@@ -138,6 +138,10 @@ class requests {
         return _finished_requests;
     }
 
+    // Returns true if there exists an inflight or finished request that covers
+    // the provided sequence range.
+    bool has_request_for_seq_range(seq_t first, seq_t last) const;
+
 private:
     bool is_valid_sequence(seq_t incoming) const;
     std::optional<request_ptr> last_request() const;

src/v/cluster/rm_stm.cc

@@ -2299,7 +2299,7 @@ ss::future<> rm_stm::apply_raft_snapshot(const iobuf&) {
     co_return;
 }
 
-bool rm_stm::is_last_batch_for_idempotent_producer(
+bool rm_stm::is_batch_in_idempotent_window(
   const model::record_batch_header& hdr) const {
     const auto bid = model::batch_identity::from(hdr);
     if (!bid.is_idempotent()) {
@@ -2310,18 +2310,20 @@ bool rm_stm::is_last_batch_for_idempotent_producer(
 
     auto it = _producers.find(pid.get_id());
     if (it == _producers.end()) {
-        // We cannot know for sure if this is the last batch for the
-        // producer or not. But we cannot retain placeholder batches forever
-        // either.
+        // We cannot know for sure if this batch is in the idempotent window
+        // or not. But we cannot retain placeholder batches forever either.
         return false;
     }
 
     const tx::producer_ptr& producer = it->second;
+    if (producer->id().get_epoch() != pid.get_epoch()) {
+        return false;
+    }
 
-    const auto last_seq = producer->last_sequence_number();
-    const auto producer_epoch = producer->id().get_epoch();
-
-    return last_seq == bid.last_seq && producer_epoch == pid.get_epoch();
+    // Check if the batch matches any request in the idempotent window
+    // (inflight or finished requests).
+    return producer->idempotent_request_state().has_request_for_seq_range(
+      bid.first_seq, bid.last_seq);
 }
 
 void rm_stm::setup_metrics() {

src/v/cluster/rm_stm.h

@@ -225,7 +225,7 @@ class rm_stm final : public raft::persisted_stm<> {
         return raft::stm_initial_recovery_policy::read_everything;
     }
 
-    bool is_last_batch_for_idempotent_producer(
+    bool is_batch_in_idempotent_window(
       const model::record_batch_header&) const override;
 
 protected:

src/v/cluster/tests/producer_state_tests.cc

@@ -558,3 +558,40 @@ FIXTURE_TEST(test_transaction_start_without_fence, test_fixture) {
       another_tx_state->status, partition_transaction_status::ongoing);
     BOOST_REQUIRE_EQUAL(another_tx_state->first, another_batch.base_offset());
 }
+
+FIXTURE_TEST(test_has_request_for_seq_range, test_fixture) {
+    create_producer_state_manager(1, 1);
+    auto producer = new_producer();
+    auto defer = ss::defer(
+      [&] { manager().deregister_producer(*producer, std::nullopt); });
+
+    model::test::record_batch_spec spec{
+      .offset = model::offset{10},
+      .allow_compression = true,
+      .count = 5,
+      .bt = model::record_batch_type::raft_data,
+      .enable_idempotence = true,
+      .producer_id = producer->id().id,
+      .producer_epoch = producer->id().epoch,
+      .base_sequence = 0,
+    };
+    auto batch = model::test::make_random_batch(spec);
+    auto bid = model::batch_identity::from(batch.header());
+    auto request = producer->try_emplace_request(bid, model::term_id{1}, true);
+    BOOST_REQUIRE(!request.has_error());
+
+    const auto& reqs = producer->idempotent_request_state();
+
+    // Inflight request with seq range [0, 4] should be found.
+    BOOST_REQUIRE(reqs.has_request_for_seq_range(0, 4));
+    // Non-matching ranges should not be found.
+    BOOST_REQUIRE(!reqs.has_request_for_seq_range(0, 3));
+    BOOST_REQUIRE(!reqs.has_request_for_seq_range(1, 4));
+    BOOST_REQUIRE(!reqs.has_request_for_seq_range(5, 9));
+
+    // Apply the batch to promote it to finished.
+    producer->apply_data(batch.header(), kafka::offset{10});
+
+    // Still found after apply - now in finished requests.
+    BOOST_REQUIRE(reqs.has_request_for_seq_range(0, 4));
+}

src/v/cluster/tests/rm_stm_tests.cc

@@ -1006,8 +1006,7 @@ FIXTURE_TEST(test_tx_compaction_last_producer_batch, rm_stm_test_fixture) {
             if (
               b.header().type == model::record_batch_type::raft_data
               && !b.header().attrs.is_control()) {
-                BOOST_REQUIRE(
-                  stm.is_last_batch_for_idempotent_producer(b.header()));
+                BOOST_REQUIRE(stm.is_batch_in_idempotent_window(b.header()));
             }
         }
     }
@@ -1025,8 +1024,9 @@ FIXTURE_TEST(test_tx_compaction_last_producer_batch, rm_stm_test_fixture) {
                          std::move(rdr), model::no_timeout)
                          .get();
 
-        // Expect the last non-control raft data batch to be the last batch for
-        // a producer.
+        // Expect the last non-control raft data batch to be in the idempotent
+        // window. The first batch is no longer in the window because
+        // begin_tx resets the request state.
         bool seen_first_batch = false;
         bool seen_last_batch = false;
         for (const auto& b : batches) {
@@ -1035,11 +1035,11 @@ FIXTURE_TEST(test_tx_compaction_last_producer_batch, rm_stm_test_fixture) {
               && !b.header().attrs.is_control()) {
                 if (seen_first_batch) {
                     BOOST_REQUIRE(
-                      stm.is_last_batch_for_idempotent_producer(b.header()));
+                      stm.is_batch_in_idempotent_window(b.header()));
                     seen_last_batch = true;
                 } else {
                     BOOST_REQUIRE(
-                      !stm.is_last_batch_for_idempotent_producer(b.header()));
+                      !stm.is_batch_in_idempotent_window(b.header()));
                     seen_first_batch = true;
                 }
             }
@@ -1082,8 +1082,7 @@ FIXTURE_TEST(test_tx_compaction_last_producer_batch, rm_stm_test_fixture) {
             if (
               b.header().type == model::record_batch_type::raft_data
               && !b.header().attrs.is_control()) {
-                BOOST_REQUIRE(
-                  stm.is_last_batch_for_idempotent_producer(b.header()));
+                BOOST_REQUIRE(stm.is_batch_in_idempotent_window(b.header()));
                 ++num_seen_raft_batches;
             }
         }
@@ -1119,8 +1118,7 @@ FIXTURE_TEST(test_tx_compaction_last_producer_batch, rm_stm_test_fixture) {
             if (
               b.header().type
               == model::record_batch_type::compaction_placeholder) {
-                BOOST_REQUIRE(
-                  stm.is_last_batch_for_idempotent_producer(b.header()));
+                BOOST_REQUIRE(stm.is_batch_in_idempotent_window(b.header()));
                 if (b.header().producer_id == pid_zero.get_id()()) {
                     seen_placeholder_batch_pid_zero = true;
                 }

src/v/storage/compaction_reducers.cc

@@ -161,7 +161,7 @@ copy_data_segment_reducer::filter(model::record_batch batch) {
     if (
       (batch.header().type == model::record_batch_type::compaction_placeholder)
       && !is_last_batch_in_segment
-      && !_stm_mgr->is_last_batch_for_idempotent_producer(batch.header())) {
+      && !_stm_mgr->is_batch_in_idempotent_window(batch.header())) {
         co_return std::nullopt;
     }
 
@@ -183,9 +183,9 @@ copy_data_segment_reducer::filter(model::record_batch batch) {
       });
 
     if (offset_deltas.empty() && _compaction_placeholder_enabled) {
-        auto is_last_batch_for_producer
-          = _stm_mgr->is_last_batch_for_idempotent_producer(batch.header());
-        if (is_last_batch_in_segment || is_last_batch_for_producer) {
+        auto is_batch_in_idempotent_window
+          = _stm_mgr->is_batch_in_idempotent_window(batch.header());
+        if (is_last_batch_in_segment || is_batch_in_idempotent_window) {
             // last batch in the segment or for the producer has been compacted
             // away. This is most likely caused by aborted data batches getting
             // compacted away during self compaction of the segment if they are

src/v/storage/types.cc

@@ -70,12 +70,12 @@ void stm_manager::set_max_tx_end_remove_offset(model::offset o) {
     _max_tx_end_remove_offset = o;
 }
 
-bool stm_manager::is_last_batch_for_idempotent_producer(
+bool stm_manager::is_batch_in_idempotent_window(
   const model::record_batch_header& hdr) const {
     if (!_tx_stm) {
         return false;
     }
-    return _tx_stm->is_last_batch_for_idempotent_producer(hdr);
+    return _tx_stm->is_batch_in_idempotent_window(hdr);
 }
 
 fmt::iterator local_log_reader_config::format_to(fmt::iterator it) const {

src/v/storage/types.h

@@ -92,8 +92,8 @@ class snapshotable_stm {
         return model::control_record_type::unknown;
     }
 
-    virtual bool is_last_batch_for_idempotent_producer(
-      const model::record_batch_header&) const {
+    virtual bool
+    is_batch_in_idempotent_window(const model::record_batch_header&) const {
         return false;
     }
 };
@@ -197,8 +197,7 @@ class stm_manager {
      */
     model::offset tx_snapshot_offset() const;
 
-    bool is_last_batch_for_idempotent_producer(
-      const model::record_batch_header&) const;
+    bool is_batch_in_idempotent_window(const model::record_batch_header&) const;
 
 private:
     ss::shared_ptr<snapshotable_stm> _tx_stm;