cl/service: update consumer configuration when link configuration change

When link configuration changes we must update configuration of consumer
used to fetch data from the source cluster.

This commits register manager notification that listens for the link
configuration changes and updates the mux consumer configuration.

Signed-off-by: Michał Maślanka <[email protected]>

Author: mmaslankaprv

src/v/cluster_link/service.cc

@@ -217,19 +217,40 @@ class remote_partition_source : public replication::data_source {
 class remote_data_source_factory : public replication::data_source_factory {
 public:
     explicit remote_data_source_factory(
+      model::id_t link_id,
+      manager* manager,
       std::unique_ptr<replication::mux_remote_consumer> consumer)
-      : _consumer(std::move(consumer)) {}
+      : _link_id(link_id)
+      , _manager(manager)
+      , _consumer(std::move(consumer)) {}
 
-    ss::future<> start() final { return _consumer->start(); }
+    ss::future<> start() final {
+        _notification_id = _manager->register_link_config_changes_callback(
+          [this](model::id_t link_id, const model::metadata& md) {
+              // Ignore updates for other links
+              if (link_id != _link_id) {
+                  return;
+              }
+              _consumer->update_configuration(
+                make_remote_consumer_configuration(md.connection));
+          });
+        return _consumer->start();
+    }
 
-    ss::future<> stop() noexcept final { return _consumer->stop(); }
+    ss::future<> stop() noexcept final {
+        _manager->unregister_link_config_changes_callback(_notification_id);
+        return _consumer->stop();
+    }
 
     std::unique_ptr<replication::data_source>
     make_source(const ::model::ntp& ntp) final {
         return make_default_data_source(ntp.tp, *_consumer);
     }
 
 private:
+    model::id_t _link_id;
+    manager* _manager;
+    manager::notification_id _notification_id;
     std::unique_ptr<replication::mux_remote_consumer> _consumer;
 };
 
@@ -406,6 +427,8 @@ class default_link_factory : public link_factory {
           std::move(config),
           std::move(cluster_connection),
           std::make_unique<remote_data_source_factory>(
+            link_id,
+            manager,
             std::make_unique<replication::mux_remote_consumer>(
               *cluster_connection,
               make_remote_consumer_configuration(config.connection))),