CORE-13396 pass client configuration to consumer

proto/redpanda/core/admin/v2/shadow_link.proto

@@ -199,14 +199,17 @@ message ShadowLinkClientOptions {
     // If 0 is provided, defaults to 100ms
     int32 retry_backoff_ms = 8;
     // Fetch request timeout
-    // If 0 is provided, defaults to 100ms
+    // If 0 is provided, defaults to 500ms
     int32 fetch_wait_max_ms = 9;
     // Fetch min bytes
-    // If 0 is provided, defaults to 1 byte
+    // If 0 is provided, defaults to 5 MiB
     int32 fetch_min_bytes = 10;
     // Fetch max bytes
-    // If 0 is provided, defaults to 1MiB
+    // If 0 is provided, defaults to 20 MiB
     int32 fetch_max_bytes = 11;
+    // Fetch partition max bytes
+    // If 0 is provided, defaults to 1 MiB
+    int32 fetch_partition_max_bytes = 12;
 }
 // Options for syncing topic metadata
 message TopicMetadataSyncOptions {

src/v/cluster_link/BUILD

@@ -193,10 +193,11 @@ 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:deps",
         "//src/v/cluster_link/replication:mux_remote_consumer",
         "//src/v/kafka/client/direct_consumer",
         "//src/v/kafka/server",
+        "//src/v/kafka/server:write_at_offset_stm",
         "//src/v/model",
         "//src/v/raft",
         "@seastar",

src/v/cluster_link/manager.cc

@@ -357,6 +357,7 @@ ss::future<> manager::handle_on_link_change(model::id_t id) {
           id,
           link_metadata);
         it->second->update_config(link_metadata.copy());
+        _cfg_change_notifications.notify(id, link_metadata);
     } else {
         // Create a new link
         vlog(
@@ -620,6 +621,15 @@ ss::future<> manager::stop_topic_reconciler() {
     }
 }
 
+manager::notification_id manager::register_link_config_changes_callback(
+  link_cfg_change_notification_cb cb) {
+    return _cfg_change_notifications.register_cb(std::move(cb));
+}
+
+void manager::unregister_link_config_changes_callback(notification_id id) {
+    _cfg_change_notifications.unregister_cb(id);
+}
+
 consumer_groups_router& manager::get_group_router() noexcept {
     return *_group_router;
 }

src/v/cluster_link/manager.h

@@ -35,6 +35,9 @@ namespace cluster_link {
  */
 class manager {
 public:
+    using notification_id = named_type<uint32_t, struct mgr_notification_tag>;
+    using link_cfg_change_notification_cb
+      = ss::noncopyable_function<void(model::id_t, const model::metadata&)>;
     manager(
       ::model::node_id self,
       std::unique_ptr<kafka::data::rpc::partition_leader_cache>
@@ -157,6 +160,10 @@ class manager {
         return _scheduling_group;
     }
 
+    notification_id
+    register_link_config_changes_callback(link_cfg_change_notification_cb cb);
+    void unregister_link_config_changes_callback(notification_id cb);
+
 private:
     /// Called periodically to reconcile registered tasks on created links
     ss::future<> link_task_reconciler();
@@ -182,6 +189,8 @@ class manager {
 
     chunked_vector<std::unique_ptr<task_factory>> _task_factories;
     absl::flat_hash_map<id_t, std::unique_ptr<link>> _links;
+    notification_list<link_cfg_change_notification_cb, notification_id>
+      _cfg_change_notifications;
 
     ss::lowres_clock::duration _task_reconciler_interval;
     mutex _link_task_reconciler_mutex{

src/v/cluster_link/model/types.h

@@ -133,7 +133,7 @@ std::ostream& operator<<(std::ostream& os, const tls_file_or_value& t);
  */
 struct connection_config
   : serde::
-      envelope<connection_config, serde::version<0>, serde::compat_version<0>> {
+      envelope<connection_config, serde::version<1>, serde::compat_version<0>> {
     /// List of addresses to bootstrap the connection
     std::vector<net::unresolved_address> bootstrap_servers;
     /// Support authn variants.  Currently only SCRAM but update this to add
@@ -163,16 +163,20 @@ struct connection_config
     static constexpr auto retry_backoff_ms_default = 100;
     // Maximum fetch wait time
     std::optional<int32_t> fetch_wait_max_ms;
-    // Default value for fetch_wait_max_ms (100ms)
-    static constexpr auto fetch_wait_max_ms_default = 100;
+    // Default value for fetch_wait_max_ms (500ms)
+    static constexpr auto fetch_wait_max_ms_default = 500;
     // Minimum number of bytes to fetch
     std::optional<int32_t> fetch_min_bytes;
-    // Default minimum number of bytes to fetch (1B)
-    static constexpr auto fetch_min_bytes_default = 1;
+    // Default minimum number of bytes to fetch (5MiB)
+    static constexpr auto fetch_min_bytes_default = 5_MiB;
     // Maximum number of bytes to fetch
     std::optional<int32_t> fetch_max_bytes;
+    // Maximum number of bytes to fetch per partition
+    std::optional<int32_t> fetch_partition_max_bytes;
+    // Default maximum number of bytes to fetch per partition
+    static constexpr auto default_fetch_partition_max_bytes = 1_MiB;
     // Default maximum number of bytes to fetch (1MiB)
-    static constexpr auto fetch_max_bytes_default = 1 * 1024 * 1024;
+    static constexpr auto fetch_max_bytes_default = 20 * 1024 * 1024;
 
     // Returns the metadata_max_age_ms value
     int32_t get_metadata_max_age_ms() const {
@@ -204,6 +208,11 @@ struct connection_config
         return fetch_max_bytes.value_or(fetch_max_bytes_default);
     }
 
+    int32_t get_fetch_partition_max_bytes() const {
+        return fetch_partition_max_bytes.value_or(
+          default_fetch_partition_max_bytes);
+    }
+
     friend bool operator==(const connection_config&, const connection_config&)
       = default;
 
@@ -220,7 +229,8 @@ struct connection_config
           retry_backoff_ms,
           fetch_wait_max_ms,
           fetch_min_bytes,
-          fetch_max_bytes);
+          fetch_max_bytes,
+          fetch_partition_max_bytes);
     }
 
     friend std::ostream&

src/v/cluster_link/replication/BUILD

@@ -22,7 +22,10 @@ redpanda_cc_library(
     hdrs = [
         "partition_replicator.h",
     ],
-    visibility = ["__subpackages__"],
+    visibility = [
+        "__subpackages__",
+        "//src/v/cluster_link:__subpackages__",
+    ],
     deps = [
         ":deps",
         "//src/v/cluster_link:logger",
@@ -88,30 +91,11 @@ redpanda_cc_library(
     visibility = ["//src/v/cluster_link:__subpackages__"],
     deps = [
         ":partition_data_queue",
+        "//src/v/base",
         "//src/v/cluster_link:logger",
         "//src/v/container:chunked_hash_map",
         "//src/v/kafka/client/direct_consumer",
         "//src/v/ssx:future_util",
         "@seastar",
     ],
 )
-
-redpanda_cc_library(
-    name = "deps_impl",
-    srcs = [
-        "deps_impl.cc",
-    ],
-    hdrs = [
-        "deps_impl.h",
-    ],
-    visibility = ["//src/v/cluster_link:__subpackages__"],
-    deps = [
-        ":deps",
-        ":mux_remote_consumer",
-        "//src/v/cluster",
-        "//src/v/cluster_link:logger",
-        "//src/v/kafka/server:write_at_offset_stm",
-        "//src/v/ssx:future_util",
-        "@seastar",
-    ],
-)