Merge pull request #27373 from bharathv/dr11

cluster_link/replication: add remote data source and partition sink

Author: bharathv

src/v/cluster/tests/cluster_test_fixture.h

@@ -122,7 +122,8 @@ class cluster_test_fixture {
       std::optional<cloud_storage::configuration> cloud_cfg = std::nullopt,
       bool enable_legacy_upload_mode = true,
       bool iceberg_enabled = false,
-      bool cloud_topics_enabled = false) {
+      bool cloud_topics_enabled = false,
+      bool cluster_linking_enabled = false) {
         return std::make_unique<redpanda_thread_fixture>(
           node_id,
           kafka_port,
@@ -141,7 +142,8 @@ class cluster_test_fixture {
           false,
           enable_legacy_upload_mode,
           iceberg_enabled,
-          cloud_topics_enabled);
+          cloud_topics_enabled,
+          cluster_linking_enabled);
     }
 
     void add_node(
@@ -160,7 +162,8 @@ class cluster_test_fixture {
       std::optional<cloud_storage::configuration> cloud_cfg = std::nullopt,
       bool enable_legacy_upload_mode = true,
       bool iceberg_enabled = false,
-      bool cloud_topics_enabled = false) {
+      bool cloud_topics_enabled = false,
+      bool cluster_linking_enabled = false) {
         _instances.emplace(
           node_id,
           make_redpanda_fixture(
@@ -177,7 +180,8 @@ class cluster_test_fixture {
             cloud_cfg,
             enable_legacy_upload_mode,
             iceberg_enabled,
-            cloud_topics_enabled));
+            cloud_topics_enabled,
+            cluster_linking_enabled));
     }
 
     application* get_node_application(model::node_id id) {
@@ -212,7 +216,8 @@ class cluster_test_fixture {
       std::optional<cloud_storage::configuration> cloud_cfg = std::nullopt,
       bool legacy_upload_mode_enabled = true,
       bool iceberg_enabled = false,
-      bool cloud_topics_enabled = false) {
+      bool cloud_topics_enabled = false,
+      bool cluster_linking_enabled = false) {
         std::vector<config::seed_server> seeds = {};
         if (!empty_seed_starts_cluster_val || node_id != 0) {
             seeds.push_back(
@@ -232,7 +237,8 @@ class cluster_test_fixture {
           cloud_cfg,
           legacy_upload_mode_enabled,
           iceberg_enabled,
-          cloud_topics_enabled);
+          cloud_topics_enabled,
+          cluster_linking_enabled);
         return get_node_application(node_id);
     }
 

src/v/cluster_link/BUILD

@@ -73,6 +73,8 @@ redpanda_cc_library(
         "//src/v/base",
         "//src/v/cluster",
         "//src/v/cluster_link/model",
+        "//src/v/cluster_link/replication:deps",
+        "//src/v/cluster_link/replication:link_replication_mgr",
         "//src/v/container:chunked_vector",
         "//src/v/kafka/client:cluster",
         "//src/v/kafka/data/rpc",
@@ -138,6 +140,9 @@ redpanda_cc_library(
         "//src/v/base",
         "//src/v/cluster",
         "//src/v/cluster/utils:partition_change_notifier_api",
+        "//src/v/cluster_link/replication:deps_impl",
+        "//src/v/cluster_link/replication:mux_remote_consumer",
+        "//src/v/kafka/client/direct_consumer",
         "//src/v/model",
         "//src/v/raft",
         "@seastar",

src/v/cluster_link/link.cc

@@ -13,7 +13,6 @@
 
 #include "cluster_link/logger.h"
 #include "cluster_link/manager.h"
-#include "model/namespace.h"
 #include "ssx/future-util.h"
 
 #include <seastar/coroutine/as_future.hh>
@@ -57,31 +56,40 @@ link::link(
   manager* manager,
   ss::lowres_clock::duration task_reconciler_interval,
   model::metadata config,
-  std::unique_ptr<kafka::client::cluster> cluster_connection)
+  std::unique_ptr<kafka::client::cluster> cluster_connection,
+  std::unique_ptr<replication::data_source_factory> data_source_factory,
+  std::unique_ptr<replication::data_sink_factory> data_sink_factory)
   : _self(self)
   , _link_id(link_id)
   , _manager(manager)
   , _config(std::move(config))
   , _cluster_connection(std::move(cluster_connection))
+  , _replication_mgr(
+      // todo: fix me
+      ss::default_scheduling_group(),
+      std::move(data_source_factory),
+      std::move(data_sink_factory))
   , _task_reconciler_interval(task_reconciler_interval) {}
 
 ss::future<> link::start() {
     vlog(
       cllog.info, "Starting cluster link {} ({})", _config.name, _config.uuid);
     // Allow exception to propagate to the caller
     co_await _cluster_connection->start();
+    co_await _replication_mgr.start();
     co_await run_task_reconciler();
     _task_reconciler.set_callback([this] {
         ssx::spawn_with_gate(_gate, [this] { return run_task_reconciler(); });
     });
     _task_reconciler.arm_periodic(_task_reconciler_interval);
 }
 
-ss::future<> link::stop() {
+ss::future<> link::stop() noexcept {
     vlog(
       cllog.info, "Stopping cluster link {} ({})", _config.name, _config.uuid);
     _task_reconciler.cancel();
     _as.request_abort();
+    co_await _replication_mgr.stop();
     co_await _gate.close();
 
     for (auto& [_, t] : _tasks) {
@@ -90,7 +98,16 @@ ss::future<> link::stop() {
           "Stopping task {} for cluster link {}",
           t->name(),
           _config.name);
-        auto res = co_await stop_task(t.get());
+        auto res_f = co_await ss::coroutine::as_future(stop_task(t.get()));
+        if (res_f.failed()) {
+            vlog(
+              cllog.warn,
+              "Failed to stop task {}: {}",
+              t->name(),
+              res_f.get_exception());
+            continue;
+        }
+        auto res = res_f.get();
         if (!res) {
             if (res.assume_error().code() == errc::task_not_running) {
                 // that's ok, keep going
@@ -148,15 +165,34 @@ void link::update_config(model::metadata config) {
     }
 }
 
-ss::future<>
-link::handle_on_leadership_change(::model::ntp ntp, ntp_leader is_ntp_leader) {
+ss::future<> link::handle_on_leadership_change(
+  ::model::ntp ntp,
+  ntp_leader is_ntp_leader,
+  std::optional<::model::term_id> term) {
     vlog(
       cllog.trace,
-      "Cluster link {} handling leadership change for {}: {}",
+      "Cluster link {} handling leadership change for {}: {}, term: {}",
       _config.name,
       ntp,
-      is_ntp_leader);
+      is_ntp_leader,
+      term);
 
+    const auto& mirror_topics = _config.state.mirror_topics;
+    if (mirror_topics.contains(ntp.tp.topic)) {
+        vlog(
+          cllog.debug,
+          "[{}] Leadership change event for partition {}, is_leader: {}",
+          _link_id,
+          ntp,
+          is_ntp_leader);
+        if (is_ntp_leader) {
+            vassert(
+              term, "Term must be set when leadership is assumed: {}", ntp);
+            _replication_mgr.start_replicator(ntp, *term);
+        } else {
+            _replication_mgr.stop_replicator(ntp, term);
+        }
+    }
     // todo: add debouncing here so that we do not trigger multiple
     // reconciliation loops at once.
     co_await run_task_reconciler();

src/v/cluster_link/link.h

@@ -12,6 +12,8 @@
 #pragma once
 
 #include "cluster_link/model/types.h"
+#include "cluster_link/replication/deps.h"
+#include "cluster_link/replication/link_replication_mgr.h"
 #include "cluster_link/task.h"
 #include "cluster_link/types.h"
 #include "kafka/client/cluster.h"
@@ -31,22 +33,26 @@ class link {
       manager* manager,
       ss::lowres_clock::duration task_reconciler_interval,
       model::metadata config,
-      std::unique_ptr<kafka::client::cluster> cluster_connection);
+      std::unique_ptr<kafka::client::cluster> cluster_connection,
+      std::unique_ptr<replication::data_source_factory>,
+      std::unique_ptr<replication::data_sink_factory>);
     link(const link&) = delete;
     link(link&&) = delete;
     link& operator=(const link&) = delete;
     link& operator=(link&&) = delete;
     virtual ~link() = default;
 
     virtual ss::future<> start();
-    virtual ss::future<> stop();
+    virtual ss::future<> stop() noexcept;
 
     ss::future<result<void>> register_task(task_factory*);
 
     void update_config(model::metadata);
 
-    ss::future<>
-    handle_on_leadership_change(::model::ntp ntp, ntp_leader is_ntp_leader);
+    ss::future<> handle_on_leadership_change(
+      ::model::ntp ntp,
+      ntp_leader is_ntp_leader,
+      std::optional<::model::term_id>);
 
     const model::metadata& config() const;
 
@@ -111,6 +117,7 @@ class link {
     chunked_hash_map<ss::sstring, std::unique_ptr<task>> _tasks;
     model::metadata _config;
     std::unique_ptr<kafka::client::cluster> _cluster_connection;
+    replication::link_replication_manager _replication_mgr;
 
     notification_list<task_state_change_cb, task_state_notification_id>
       _task_state_change_notifications;

src/v/cluster_link/manager.cc

@@ -103,23 +103,6 @@ ss::future<> manager::start() {
         co_await handle_on_link_change(id);
     }
 
-    auto controller_node_leader = _partition_leader_cache->get_leader_node(
-      ::model::controller_ntp);
-    if (
-      controller_node_leader.has_value()
-      && controller_node_leader.value() == _self) {
-        auto controller_shard_leader = _partition_manager->shard_owner(
-          ::model::controller_ntp);
-        if (
-          controller_shard_leader.has_value()
-          && controller_shard_leader.value() == ss::this_shard_id()) {
-            vlog(
-              cllog.info, "Cluster link manager started on controller shard");
-            handle_partition_state_change(
-              ::model::controller_ntp, ntp_leader::yes);
-        }
-    }
-
     _link_task_reconciler_timer.set_callback([this] {
         ssx::spawn_with_gate(_g, [this] { return link_task_reconciler(); });
     });
@@ -223,10 +206,12 @@ void manager::on_link_change(model::id_t id) {
 }
 
 void manager::handle_partition_state_change(
-  ::model::ntp ntp, ntp_leader is_ntp_leader) {
+  ::model::ntp ntp,
+  ntp_leader is_ntp_leader,
+  std::optional<::model::term_id> term) {
     vlog(cllog.trace, "NTP={} leadership changed to {}", ntp, is_ntp_leader);
-    _queue.submit([this, ntp{std::move(ntp)}, is_ntp_leader]() mutable {
-        return handle_on_leadership_change(std::move(ntp), is_ntp_leader);
+    _queue.submit([this, ntp{std::move(ntp)}, is_ntp_leader, term]() mutable {
+        return handle_on_leadership_change(std::move(ntp), is_ntp_leader, term);
     });
 }
 
@@ -243,20 +228,17 @@ ss::future<> manager::handle_on_link_change(model::id_t id) {
             try {
                 vlog(cllog.debug, "Stopping cluster link with id={}", id);
                 co_await it->second->stop();
-                _links.erase(it);
             } catch (const std::exception& e) {
+                // generally not possible since stop() is noexcept
+                // but is not enforced for coroutines by the compiler.
                 vlog(
                   cllog.warn,
-                  "Failed to stop link {}: \"{}\".  Re-attempting link "
-                  "stop "
-                  "within {} seconds",
+                  "Failed to stop link {}: \"{}, going ahead and removing "
+                  "it\".",
                   id,
-                  e,
-                  retry_delay.count());
-                _queue.submit_delayed(retry_delay, [this, id] {
-                    return handle_on_link_change(id);
-                });
+                  e);
             }
+            _links.erase(it);
         } else {
             vlog(cllog.trace, "No link found for id={}", id);
         }
@@ -312,7 +294,30 @@ ss::future<> manager::handle_on_link_change(model::id_t id) {
                       e);
                 }
             }
-            co_await new_link->start();
+
+            std::exception_ptr start_eptr = nullptr;
+            try {
+                co_await new_link->start();
+            } catch (...) {
+                start_eptr = std::current_exception();
+            }
+            if (start_eptr) {
+                vlog(
+                  cllog.warn,
+                  "Failed to start link {}: \"{}\"",
+                  id,
+                  start_eptr);
+                try {
+                    co_await new_link->stop();
+                } catch (...) {
+                    vlog(
+                      cllog.warn,
+                      "Failed to stop link {}: \"{}\", ignoring..",
+                      id,
+                      std::current_exception());
+                }
+                std::rethrow_exception(start_eptr);
+            }
             _links.emplace(id, std::move(new_link));
             _link_created_cv.broadcast();
         } catch (const ss::semaphore_aborted&) {
@@ -394,7 +399,9 @@ ss::future<> manager::link_task_reconciler() {
 }
 
 ss::future<> manager::handle_on_leadership_change(
-  ::model::ntp ntp, ntp_leader is_ntp_leader) {
+  ::model::ntp ntp,
+  ntp_leader is_ntp_leader,
+  std::optional<::model::term_id> term) {
     vlog(
       cllog.trace,
       "Handling leadership change for NTP={}, is_ntp_leader={}",
@@ -415,9 +422,11 @@ ss::future<> manager::handle_on_leadership_change(
         }
     }
 
-    co_await ss::parallel_for_each(_links, [ntp, is_ntp_leader](auto& pair) {
-        return pair.second->handle_on_leadership_change(ntp, is_ntp_leader);
-    });
+    co_await ss::parallel_for_each(
+      _links, [ntp, is_ntp_leader, term](auto& pair) {
+          return pair.second->handle_on_leadership_change(
+            ntp, is_ntp_leader, term);
+      });
 }
 
 ss::future<::cluster::cluster_link::errc> manager::add_mirror_topic(

src/v/cluster_link/manager.h

@@ -72,12 +72,18 @@ class manager {
     /// Used to notify that a cluster link has been updated
     void on_link_change(model::id_t id);
     /// Used to notify manager in a change of NTP leadership
-    void
-    handle_partition_state_change(::model::ntp ntp, ntp_leader is_ntp_leader);
+    void handle_partition_state_change(
+      ::model::ntp ntp,
+      ntp_leader is_ntp_leader,
+      std::optional<::model::term_id>);
     /// Handles creation and start of a link
     ss::future<> handle_on_link_change(model::id_t id);
     /// Handles leadership changes for a given NTP
-    ss::future<> handle_on_leadership_change(::model::ntp, ntp_leader);
+    /// term will be set if partition still exists on the shard
+    /// Will definitely be set if is_ntp_leader == true because assuming
+    // leadership implies the partition is still present
+    ss::future<> handle_on_leadership_change(
+      ::model::ntp, ntp_leader, std::optional<::model::term_id>);
     /// Used to add a mirror topic to a cluster link
     ss::future<::cluster::cluster_link::errc>
     add_mirror_topic(model::id_t link_id, model::add_mirror_topic_cmd cmd);