Merge pull request #27977 from nvartolomei/nv/backport-27952

Author: nvartolomei

src/v/cloud_storage/partition_manifest.cc

@@ -714,12 +714,19 @@ bool partition_manifest::advance_start_offset(model::offset new_start_offset) {
             subtract_from_cloud_log_size(it->size_bytes);
         }
 
-        // The new start offset has moved past the user-requested start offset,
-        // so there's no point in tracking it further.
-        if (
-          highest_removed_offset != kafka::offset{}
-          && highest_removed_offset >= _start_kafka_offset_override) {
-            _start_kafka_offset_override = kafka::offset{};
+        if (_archive_start_offset == model::offset{}) {
+            // This method is used by spillover (archive) mechanism to advance
+            // the start offset of the manifest and by STM retention mechanism.
+            // We should touch _start_kafka_offset_override only if the archive
+            // is not used. Otherwise, the override should be managed by the
+            // archive logic (apply_archive_retention, etc).
+            if (
+              highest_removed_offset != kafka::offset{}
+              && highest_removed_offset >= _start_kafka_offset_override) {
+                // The new start offset has moved past the user-requested start
+                // offset, so there's no point in tracking it further.
+                _start_kafka_offset_override = kafka::offset{};
+            }
         }
         return true;
     }

src/v/cloud_storage/tests/cloud_storage_e2e_test.cc

@@ -19,6 +19,7 @@
 #include "cluster/health_monitor_frontend.h"
 #include "kafka/data/replicated_partition.h"
 #include "kafka/protocol/find_coordinator.h"
+#include "kafka/server/tests/delete_records_utils.h"
 #include "kafka/server/tests/list_offsets_utils.h"
 #include "kafka/server/tests/produce_consume_utils.h"
 #include "model/fundamental.h"
@@ -1267,6 +1268,95 @@ TEST_P(EndToEndFixture, TestConsumerOffsetsNoTieredStorage) {
     ASSERT_FALSE(partition->cloud_data_available());
 }
 
+TEST_F(ManualFixture, TestSpilloverWithTruncationRetainsStartOffset) {
+    test_local_cfg.get("cloud_storage_disable_upload_loop_for_tests")
+      .set_value(true);
+    test_local_cfg.get("cloud_storage_spillover_manifest_max_segments")
+      .set_value(std::make_optional<size_t>(2));
+    test_local_cfg.get("cloud_storage_spillover_manifest_size")
+      .set_value(std::optional<size_t>{});
+
+    const model::topic topic_name("spillover_truncate_test");
+    model::ntp ntp(model::kafka_namespace, topic_name, 0);
+
+    cluster::topic_properties props;
+    props.shadow_indexing = model::shadow_indexing_mode::full;
+    props.cleanup_policy_bitflags = model::cleanup_policy_bitflags::deletion;
+    props.retention_bytes = tristate<size_t>(disable_tristate_t{});
+    props.retention_duration = tristate<std::chrono::milliseconds>(
+      disable_tristate_t{});
+    add_topic({model::kafka_namespace, topic_name}, 1, props).get();
+    wait_for_leader(ntp).get();
+
+    auto partition = app.partition_manager.local().get(ntp);
+    auto& archiver = partition->archiver().value().get();
+
+    SCOPED_TRACE("Seeding partition data");
+
+    {
+        tests::remote_segment_generator gen(
+          make_kafka_client().get(), *partition);
+        auto first_batch = gen.num_segments(6)
+                             .batches_per_segment(5)
+                             .records_per_batch(1)
+                             .produce()
+                             .get();
+        ASSERT_GE(first_batch, 30);
+    }
+
+    // Sync and upload segments
+    ASSERT_TRUE(archiver.sync_for_tests().get());
+
+    // Step 2: Apply spillover
+    vlog(e2e_test_log.info, "Applying first spillover");
+    archiver.apply_spillover().get();
+
+    // Capture start offset before truncation
+    auto start_kafka_offset_before
+      = archiver.manifest().full_log_start_kafka_offset();
+    vlog(
+      e2e_test_log.info,
+      "Start kafka offset before truncation: {}",
+      start_kafka_offset_before);
+
+    SCOPED_TRACE("Truncating via DeleteRecords");
+    const auto timeout = 10s;
+
+    tests::kafka_delete_records_transport deleter(make_kafka_client().get());
+    deleter.start().get();
+
+    // Delete offset from second segment of the first spillover manifest.
+    deleter
+      .delete_records_from_partition(
+        topic_name, ntp.tp.partition, model::offset{8}, timeout)
+      .get();
+
+    ASSERT_EQ(
+      archiver.manifest().full_log_start_kafka_offset(), kafka::offset{0});
+    ASSERT_EQ(
+      archiver.manifest().get_start_kafka_offset_override(), kafka::offset{8});
+
+    // Additional segments to trigger another spillover round.
+    {
+        tests::remote_segment_generator gen(
+          make_kafka_client().get(), *partition);
+        gen.num_segments(6)
+          .batches_per_segment(5)
+          .records_per_batch(1)
+          .produce()
+          .get();
+    }
+
+    archiver.housekeeping().get();
+
+    // Due to implementation deficiencies truncation doesn't advance start
+    // offset but at least the override should not regress.
+    ASSERT_EQ(
+      archiver.manifest().full_log_start_kafka_offset(), kafka::offset{0});
+    ASSERT_EQ(
+      archiver.manifest().get_start_kafka_offset_override(), kafka::offset{8});
+}
+
 INSTANTIATE_TEST_SUITE_P(WithOverride, EndToEndFixture, ::testing::Bool());
 
 INSTANTIATE_TEST_SUITE_P(

src/v/cluster/archival/tests/archival_metadata_stm_test.cc

@@ -905,14 +905,17 @@ FIXTURE_TEST(
       kafka::offset(1200));
     BOOST_REQUIRE_EQUAL(archival_stm->get_start_offset(), model::offset(1000));
 
-    // Advancing the start offset past the override resets the override.
+    // This does reset the override as archive is not truncated yet.
     archival_stm
       ->truncate(
         model::offset(2000), ss::lowres_clock::now() + 10s, never_abort)
       .get();
     BOOST_REQUIRE_EQUAL(
       archival_stm->manifest().get_start_kafka_offset_override(),
-      kafka::offset{});
+      kafka::offset{1200});
+
+    // This is STM start offset. Previous truncate truncated as-if applied by
+    // spillover.
     BOOST_REQUIRE_EQUAL(archival_stm->get_start_offset(), model::offset(2000));
 }