Merge pull request #25904 from Lazin/manual-backport-24566-v24.2.x

[v24.2.x] archival: Disable cross-term compaction

Author: piyushredpanda

src/v/archival/segment_reupload.cc

@@ -10,6 +10,7 @@
 
 #include "segment_reupload.h"
 
+#include "base/vlog.h"
 #include "cloud_storage/partition_manifest.h"
 #include "config/configuration.h"
 #include "logger.h"
@@ -190,6 +191,21 @@ void segment_collector::do_collect(segment_collector_mode mode) {
                result.segment->offsets().get_base_offset()});
             align_begin_offset_to_manifest();
         }
+
+        // Only segments from the same term can be concatenated together.
+        if (
+          !_segments.empty()
+          && _segments.back()->offsets().get_term()
+               != result.segment->offsets().get_term()) {
+            archival_log.debug(
+              "Segment collect for ntp {} stopping collection, last segment "
+              "term {} is different from current segment term: {}",
+              _manifest.get_ntp(),
+              _segments.back()->offsets().get_term(),
+              result.segment->offsets().get_term());
+            break;
+        }
+
         _segments.push_back(result.segment);
         current_segment_end = result.segment->offsets().get_committed_offset();
         start = current_segment_end + model::offset{1};

src/v/archival/tests/ntp_archiver_reupload_test.cc

@@ -17,6 +17,7 @@
 #include "config/configuration.h"
 #include "storage/ntp_config.h"
 #include "test_utils/fixture.h"
+#include "test_utils/scoped_config.h"
 
 #include <seastar/core/sharded.hh>
 
@@ -226,7 +227,7 @@ struct reupload_fixture : public archiver_fixture {
           manifest_view);
     }
 
-    ss::lw_shared_ptr<storage::segment> self_compact_next_segment(
+    ss::lw_shared_ptr<storage::segment> run_disk_log_housekeeping(
       model::offset max_collectible = model::offset::max()) {
         auto& seg_set = disk_log_impl()->segments();
         auto size_before = seg_set.size();
@@ -322,7 +323,7 @@ FIXTURE_TEST(test_upload_compacted_segments, reupload_fixture) {
     // Mark first segment compacted, and re-upload, now only one segment is
     // uploaded.
     reset_http_call_state();
-    auto seg = self_compact_next_segment(
+    auto seg = run_disk_log_housekeeping(
       stm_manifest.first_addressable_segment()->committed_offset);
 
     expected = archival::ntp_archiver::batch_result{{0, 0, 0}, {1, 0, 0}};
@@ -340,7 +341,7 @@ FIXTURE_TEST(test_upload_compacted_segments, reupload_fixture) {
     BOOST_REQUIRE_EQUAL(replaced[0].base_offset, model::offset{0});
 
     // Mark second segment as compacted and re-upload.
-    seg = self_compact_next_segment(
+    seg = run_disk_log_housekeeping(
       std::next(stm_manifest.first_addressable_segment())->committed_offset);
 
     reset_http_call_state();
@@ -366,7 +367,7 @@ FIXTURE_TEST(test_upload_compacted_segments, reupload_fixture) {
 FIXTURE_TEST(test_upload_compacted_segments_concat, reupload_fixture) {
     std::vector<segment_desc> segments = {
       {manifest_ntp, model::offset(0), model::term_id(1), 1000, 2},
-      {manifest_ntp, model::offset(1000), model::term_id(4), 10, 2},
+      {manifest_ntp, model::offset(1000), model::term_id(1), 10, 2},
     };
 
     initialize(segments);
@@ -384,7 +385,7 @@ FIXTURE_TEST(test_upload_compacted_segments_concat, reupload_fixture) {
 
     auto manifest = verify_manifest_request(*part);
     verify_segment_request("0-1-v1.log", manifest);
-    verify_segment_request("1000-4-v1.log", manifest);
+    verify_segment_request("1000-1-v1.log", manifest);
 
     BOOST_REQUIRE(part->archival_meta_stm());
     const cloud_storage::partition_manifest& stm_manifest
@@ -397,7 +398,7 @@ FIXTURE_TEST(test_upload_compacted_segments_concat, reupload_fixture) {
     // Mark both segments compacted, and re-upload. One concatenated segment is
     // uploaded.
     reset_http_call_state();
-    auto seg = self_compact_next_segment();
+    auto seg = run_disk_log_housekeeping();
 
     expected = archival::ntp_archiver::batch_result{{0, 0, 0}, {1, 0, 0}};
     upload_and_verify(archiver.value(), expected);
@@ -412,8 +413,7 @@ FIXTURE_TEST(test_upload_compacted_segments_concat, reupload_fixture) {
     BOOST_REQUIRE_EQUAL(replaced[0].base_offset, model::offset{0});
     BOOST_REQUIRE_EQUAL(replaced[1].base_offset, model::offset{1000});
 
-    verify_concat_segment_request(
-      {"0-1-v1.log", "1000-4-v1.log"}, part->archival_meta_stm()->manifest());
+    BOOST_REQUIRE_EQUAL(stm_manifest.size(), 1);
 }
 
 FIXTURE_TEST(
@@ -438,7 +438,7 @@ FIXTURE_TEST(
     listen();
 
     // Self-compact just the first couple segments.
-    self_compact_next_segment(model::offset{999});
+    run_disk_log_housekeeping(model::offset{999});
 
     archival::ntp_archiver::batch_result expected{{0, 0, 0}, {1, 0, 0}};
     upload_and_verify(archiver.value(), expected);
@@ -475,7 +475,7 @@ FIXTURE_TEST(test_upload_compacted_segments_fill_gap, reupload_fixture) {
 
     listen();
 
-    self_compact_next_segment();
+    run_disk_log_housekeeping();
 
     archival::ntp_archiver::batch_result expected{{0, 0, 0}, {1, 0, 0}};
     upload_and_verify(archiver.value(), expected);
@@ -541,7 +541,7 @@ FIXTURE_TEST(test_upload_both_compacted_and_non_compacted, reupload_fixture) {
     // NOTE: we can only compact up to what's been uploaded, since that
     // determines the max collectible offset.
     reset_http_call_state();
-    auto seg = self_compact_next_segment(
+    auto seg = run_disk_log_housekeeping(
       manifest.first_addressable_segment()->committed_offset);
 
     expected = archival::ntp_archiver::batch_result{{1, 0, 0}, {1, 0, 0}};
@@ -607,7 +607,7 @@ FIXTURE_TEST(test_both_uploads_with_one_failing, reupload_fixture) {
     // Self-compact the first segment and re-upload. One compacted
     // and one non-compacted segments are uploaded.
     reset_http_call_state();
-    auto seg = self_compact_next_segment(disk_log_impl()
+    auto seg = run_disk_log_housekeeping(disk_log_impl()
                                            ->segments()
                                            .begin()
                                            ->get()
@@ -671,11 +671,9 @@ FIXTURE_TEST(test_upload_when_compaction_disabled, reupload_fixture) {
     BOOST_REQUIRE_EQUAL(
       stm_manifest.get_last_uploaded_compacted_offset(), model::offset{});
 
-    // Self-compact the first segment, since the topic has compaction
-    // disabled, and re-upload, nothing is uploaded.
+    // Since the topic has compaction is disabled nothing should
+    // be reuploaded
     reset_http_call_state();
-    auto seg = self_compact_next_segment();
-
     expected = archival::ntp_archiver::batch_result{{0, 0, 0}, {0, 0, 0}};
     upload_and_verify(archiver.value(), expected);
     BOOST_REQUIRE_EQUAL(get_requests().size(), 0);
@@ -719,7 +717,7 @@ FIXTURE_TEST(test_upload_when_reupload_disabled, reupload_fixture) {
     // Mark first segment compacted artificially, since the topic has compaction
     // disabled, and re-upload, nothing is uploaded.
     reset_http_call_state();
-    auto seg = self_compact_next_segment();
+    auto seg = run_disk_log_housekeeping();
 
     expected = archival::ntp_archiver::batch_result{{0, 0, 0}, {0, 0, 0}};
 
@@ -745,10 +743,10 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
     // NOTE: different terms so compaction leaves one segment each.
     std::vector<segment_desc> segments = {
       {manifest_ntp, model::offset(0), model::term_id(1), 10, 2},
-      {manifest_ntp, model::offset(10), model::term_id(2), 10, 2},
-      {manifest_ntp, model::offset(20), model::term_id(3), 10, 2},
-      {manifest_ntp, model::offset(30), model::term_id(4), 10, 2},
-      {manifest_ntp, model::offset(40), model::term_id(5), 10, 2},
+      {manifest_ntp, model::offset(10), model::term_id(1), 10, 2},
+      {manifest_ntp, model::offset(20), model::term_id(1), 10, 2},
+      {manifest_ntp, model::offset(30), model::term_id(1), 10, 2},
+      {manifest_ntp, model::offset(40), model::term_id(1), 10, 2},
     };
 
     initialize(segments);
@@ -767,9 +765,9 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
       get_targets().find(manifest_url)->second.content);
 
     verify_segment_request("0-1-v1.log", manifest);
-    verify_segment_request("10-2-v1.log", manifest);
-    verify_segment_request("20-3-v1.log", manifest);
-    verify_segment_request("30-4-v1.log", manifest);
+    verify_segment_request("10-1-v1.log", manifest);
+    verify_segment_request("20-1-v1.log", manifest);
+    verify_segment_request("30-1-v1.log", manifest);
 
     BOOST_REQUIRE(part->archival_meta_stm());
     const cloud_storage::partition_manifest& stm_manifest
@@ -789,28 +787,28 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
         create_segment(
           {manifest_ntp,
            last_segment->offsets().get_committed_offset() + model::offset{1},
-           model::term_id{6},
+           model::term_id{2},
            10});
     }
 
     reset_http_call_state();
 
     // Mark four segments as compacted, so they are valid for upload
     ss::lw_shared_ptr<storage::segment> seg;
-    seg = self_compact_next_segment(model::offset(39));
+    seg = run_disk_log_housekeeping(model::offset(39));
 
     expected = archival::ntp_archiver::batch_result{{4, 0, 0}, {0, 0, 0}};
     upload_and_verify(archiver.value(), expected, model::offset::max());
     BOOST_REQUIRE_EQUAL(get_requests().size(), 9);
 
     verify_segment_request(
-      "40-5-v1.log", part->archival_meta_stm()->manifest());
+      "40-1-v1.log", part->archival_meta_stm()->manifest());
     verify_segment_request(
-      "50-6-v1.log", part->archival_meta_stm()->manifest());
+      "50-2-v1.log", part->archival_meta_stm()->manifest());
     verify_segment_request(
-      "65-6-v1.log", part->archival_meta_stm()->manifest());
+      "65-2-v1.log", part->archival_meta_stm()->manifest());
     verify_segment_request(
-      "85-6-v1.log", part->archival_meta_stm()->manifest());
+      "85-2-v1.log", part->archival_meta_stm()->manifest());
 
     BOOST_REQUIRE_EQUAL(
       stm_manifest.get_last_uploaded_compacted_offset(), model::offset{});
@@ -824,19 +822,81 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
     upload_and_verify(archiver.value(), expected);
     BOOST_REQUIRE_EQUAL(get_requests().size(), 3);
 
-    verify_concat_segment_request(
-      {
-        "0-1-v1.log",
-        "10-2-v1.log",
-        "20-3-v1.log",
-        "30-4-v1.log",
-      },
-      part->archival_meta_stm()->manifest());
-
     BOOST_REQUIRE_EQUAL(
       stm_manifest.get_last_uploaded_compacted_offset(),
       seg->offsets().get_committed_offset());
 
     replaced = stm_manifest.replaced_segments();
     BOOST_REQUIRE_EQUAL(replaced.size(), 4);
+    BOOST_REQUIRE_EQUAL(stm_manifest.size(), 5);
+}
+
+FIXTURE_TEST(test_upload_compacted_segments_cross_term, reupload_fixture) {
+    std::vector<segment_desc> segments = {
+      {manifest_ntp, model::offset(0), model::term_id(1), 1000, 2},
+      {manifest_ntp, model::offset(1000), model::term_id(4), 10, 2},
+    };
+
+    initialize(segments);
+    auto action = ss::defer([this] { archiver->stop().get(); });
+
+    auto part = app.partition_manager.local().get(manifest_ntp);
+    listen();
+
+    // Upload two non compacted segments, no segment is compacted yet.
+    archival::ntp_archiver::batch_result expected{{2, 0, 0}, {0, 0, 0}};
+    upload_and_verify(archiver.value(), expected);
+
+    // Two segments, two indices, one manifest
+    BOOST_REQUIRE_EQUAL(get_requests().size(), 5);
+
+    auto manifest = verify_manifest_request(*part);
+    verify_segment_request("0-1-v1.log", manifest);
+    verify_segment_request("1000-4-v1.log", manifest);
+
+    BOOST_REQUIRE(part->archival_meta_stm());
+    const cloud_storage::partition_manifest& stm_manifest
+      = part->archival_meta_stm()->manifest();
+    verify_stm_manifest(stm_manifest, segments);
+
+    BOOST_REQUIRE_EQUAL(
+      stm_manifest.get_last_uploaded_compacted_offset(), model::offset{});
+
+    // Mark both segments compacted, and re-upload. Both segments are
+    // re-uploaded.
+    reset_http_call_state();
+
+    vlog(test_log.info, "Waiting for segments to self-compact");
+    auto seg = run_disk_log_housekeeping();
+    vlog(test_log.info, "Self-compaction completed");
+
+    expected = archival::ntp_archiver::batch_result{{0, 0, 0}, {2, 0, 0}};
+    upload_and_verify(archiver.value(), expected);
+    BOOST_REQUIRE_EQUAL(get_requests().size(), 5);
+
+    BOOST_REQUIRE_EQUAL(
+      stm_manifest.get_last_uploaded_compacted_offset(),
+      seg->offsets().get_committed_offset());
+
+    auto replaced = stm_manifest.replaced_segments();
+    BOOST_REQUIRE_EQUAL(replaced.size(), 2);
+    BOOST_REQUIRE_EQUAL(replaced[0].base_offset, model::offset{0});
+    BOOST_REQUIRE_EQUAL(replaced[1].base_offset, model::offset{1000});
+
+    // We can't reupload x-term so we should end up with two
+    // compacted uploads.
+
+    {
+        auto it = stm_manifest.get(model::offset(0));
+        BOOST_REQUIRE_EQUAL(it->base_offset, model::offset(0));
+        BOOST_REQUIRE_EQUAL(it->committed_offset, model::offset(999));
+        BOOST_REQUIRE_EQUAL(it->segment_term, model::term_id(1));
+    }
+
+    {
+        auto it = stm_manifest.get(model::offset(1000));
+        BOOST_REQUIRE_EQUAL(it->base_offset, model::offset(1000));
+        BOOST_REQUIRE_EQUAL(it->committed_offset, model::offset(1009));
+        BOOST_REQUIRE_EQUAL(it->segment_term, model::term_id(4));
+    }
 }