Merge pull request #29802 from andrwng/set-start-no-op

ct/housekeeper: don't RPC if start offset hasn't changed

Author: andrwng

src/v/cloud_topics/housekeeper/housekeeper.cc

@@ -190,6 +190,9 @@ ss::future<> housekeeper::sync_start_offset() {
         co_return;
     }
     auto start_offset = _l0_metastore->get_start_offset(_tidp);
+    if (_last_synced_start_offset == start_offset) {
+        co_return;
+    }
     vlog(cd_log.debug, "Setting {} start offset to {}", _tidp, start_offset);
     auto result = co_await _l1_metastore->set_start_offset(_tidp, start_offset);
     if (!result.has_value()) {
@@ -198,7 +201,9 @@ ss::future<> housekeeper::sync_start_offset() {
           "Failed to sync start offset to L1 for {}: {}",
           _tidp,
           result.error());
+        co_return;
     }
+    _last_synced_start_offset = start_offset;
 }
 
 } // namespace cloud_topics

src/v/cloud_topics/housekeeper/housekeeper.h

@@ -144,6 +144,7 @@ class housekeeper {
     ss::abort_source _as;
 
     std::optional<cloud_topics::cluster_epoch> _last_epoch;
+    std::optional<kafka::offset> _last_synced_start_offset;
 };
 
 } // namespace cloud_topics

src/v/cloud_topics/housekeeper/tests/housekeeper_test.cc

@@ -197,6 +197,17 @@ T handle_error(
         std::to_underlying(result.error())));
 }
 
+class counting_metastore : public cloud_topics::l1::simple_metastore {
+public:
+    ss::future<std::expected<void, errc>> set_start_offset(
+      const model::topic_id_partition& tidp, kafka::offset o) override {
+        ++set_start_offset_calls;
+        return simple_metastore::set_start_offset(tidp, o);
+    }
+
+    size_t set_start_offset_calls{0};
+};
+
 } // namespace
 
 using namespace std::chrono_literals;
@@ -224,6 +235,10 @@ class HousekeeperTest : public testing::Test {
         _l0_metastore.set_max_allowed_start_offset(offset);
     }
 
+    size_t l1_set_start_offset_calls() const {
+        return _l1_metastore.set_start_offset_calls;
+    }
+
     kafka::offset l1_start_offset() {
         auto result = _l1_metastore.get_offsets(_tidp).get();
         if (!result.has_value()) {
@@ -289,7 +304,7 @@ class HousekeeperTest : public testing::Test {
     model::topic_id_partition _tidp{
       model::create_topic_id(), model::partition_id{0}};
     fake_l0_metastore _l0_metastore{_tidp, kafka::offset{0}};
-    cloud_topics::l1::simple_metastore _l1_metastore;
+    counting_metastore _l1_metastore;
     std::unique_ptr<retention_config_impl> _config_impl;
 };
 
@@ -848,3 +863,21 @@ TEST_F(HousekeeperTest, BumpEpochMultipleIdleCycles) {
     EXPECT_EQ(advance_epoch_calls()[0], cloud_topics::cluster_epoch{10});
     EXPECT_EQ(sync_to_next_placeholder_calls(), 1);
 }
+
+TEST_F(HousekeeperTest, SyncStartOffsetSkipsRedundantRPC) {
+    auto housekeeper = make_housekeeper({});
+    add_object({.records = 100, .size = 1_MiB});
+    set_start_offset(kafka::offset{50});
+
+    housekeeper.do_housekeeping().get();
+    EXPECT_EQ(l1_set_start_offset_calls(), 1);
+
+    // No change — should skip the call.
+    housekeeper.do_housekeeping().get();
+    EXPECT_EQ(l1_set_start_offset_calls(), 1);
+
+    // Changed offset — should sync again.
+    set_start_offset(kafka::offset{75});
+    housekeeper.do_housekeeping().get();
+    EXPECT_EQ(l1_set_start_offset_calls(), 2);
+}