[v23.1.x] archival: don't instantiate merger on read replicas

src/v/archival/adjacent_segment_merger.cc

@@ -43,7 +43,12 @@ adjacent_segment_merger::adjacent_segment_merger(
   , _target_segment_size(
       config::shard_local_cfg().cloud_storage_segment_size_target.bind())
   , _min_segment_size(
-      config::shard_local_cfg().cloud_storage_segment_size_min.bind()) {}
+      config::shard_local_cfg().cloud_storage_segment_size_min.bind()) {
+    vassert(
+      !_archiver.ntp_config().is_read_replica_mode_enabled(),
+      "Constructed adjacent segment merger on read replica {}",
+      _archiver.get_ntp());
+}
 
 ss::future<> adjacent_segment_merger::stop() { return _gate.close(); }
 
@@ -183,4 +188,4 @@ bool adjacent_segment_merger::interrupted() const {
     return _as.abort_requested();
 }
 
-} // namespace archival
+} // namespace archival

src/v/archival/ntp_archiver_service.cc

@@ -56,7 +56,9 @@ static std::unique_ptr<adjacent_segment_merger>
 maybe_make_adjacent_segment_merger(
   ntp_archiver& self, retry_chain_logger& log, const storage::ntp_config& cfg) {
     std::unique_ptr<adjacent_segment_merger> result = nullptr;
-    if (cfg.is_archival_enabled() && !cfg.is_compacted()) {
+    if (
+      cfg.is_archival_enabled() && !cfg.is_compacted()
+      && !cfg.is_read_replica_mode_enabled()) {
         result = std::make_unique<adjacent_segment_merger>(self, log, true);
         result->set_enabled(config::shard_local_cfg()
                               .cloud_storage_enable_segment_merging.value());
@@ -100,9 +102,7 @@ ntp_archiver::ntp_archiver(
     // Override bucket for read-replica
     if (_parent.is_read_replica_mode_enabled()) {
         _bucket_override = _parent.get_read_replica_bucket();
-    }
-
-    if (
+    } else if (
       parent.log().config().is_archival_enabled()
       && !parent.log().config().is_compacted()) {
         _segment_merging_enabled.watch([this] {

src/v/archival/ntp_archiver_service.h

@@ -234,6 +234,10 @@ class ntp_archiver {
     /// it will resume uploads.
     void complete_transfer_leadership();
 
+    const storage::ntp_config& ntp_config() const {
+        return _parent.log().config();
+    }
+
 private:
     ss::future<bool> do_upload_local(
       upload_candidate_with_locks candidate,

src/v/cluster/partition.cc

@@ -494,15 +494,18 @@ void partition::maybe_construct_archiver() {
     // NOTE: construct and archiver even if shadow indexing isn't enabled, e.g.
     // in the case of read replicas -- we still need the archiver to drive
     // manifest updates, etc.
+    auto& ntp_config = _raft->log().config();
     if (
       config::shard_local_cfg().cloud_storage_enabled()
       && _cloud_storage_api.local_is_initialized()
       && _raft->ntp().ns == model::kafka_namespace
-      && (_raft->log().config().is_archival_enabled() || _raft->log().config().is_read_replica_mode_enabled())) {
+      && (ntp_config.is_archival_enabled() || ntp_config.is_read_replica_mode_enabled())) {
         _archiver = std::make_unique<archival::ntp_archiver>(
-          log().config(), _archival_conf, _cloud_storage_api.local(), *this);
-        _upload_housekeeping.local().register_jobs(
-          _archiver->get_housekeeping_jobs());
+          ntp_config, _archival_conf, _cloud_storage_api.local(), *this);
+        if (!ntp_config.is_read_replica_mode_enabled()) {
+            _upload_housekeeping.local().register_jobs(
+              _archiver->get_housekeeping_jobs());
+        }
     }
 }