[v24.2.x] Fix dissemination large shard count

src/v/cluster/metadata_dissemination_handler.cc

@@ -19,9 +19,11 @@
 #include "model/fundamental.h"
 #include "model/metadata.h"
 #include "model/timeout_clock.h"
+#include "ssx/async_algorithm.h"
 
 #include <seastar/core/chunked_fifo.hh>
 #include <seastar/core/loop.hh>
+#include <seastar/core/shard_id.hh>
 #include <seastar/core/smp.hh>
 
 #include <boost/range/irange.hpp>
@@ -49,29 +51,28 @@ metadata_dissemination_handler::update_leadership_v2(
 
 ss::future<update_leadership_reply>
 metadata_dissemination_handler::do_update_leadership(
-  fragmented_vector<ntp_leader_revision> leaders) {
+  chunked_vector<ntp_leader_revision> leaders) {
     vlog(clusterlog.trace, "Received a metadata update");
-    co_await ss::parallel_for_each(
-      boost::irange<ss::shard_id>(0, ss::smp::count),
-      [this, leaders = std::move(leaders)](ss::shard_id shard) {
-          fragmented_vector<ntp_leader_revision> local_leaders;
-          local_leaders.reserve(leaders.size());
-          std::copy(
-            leaders.begin(), leaders.end(), std::back_inserter(local_leaders));
-
-          return ss::smp::submit_to(
-            shard, [this, leaders = std::move(local_leaders)] {
-                for (auto& leader : leaders) {
-                    _leaders.local().update_partition_leader(
-                      leader.ntp,
-                      leader.revision,
-                      leader.term,
-                      leader.leader_id);
-                }
-            });
-      });
-
-    co_return update_leadership_reply{};
+    return ss::do_with(
+             std::move(leaders),
+             [this](const chunked_vector<ntp_leader_revision>& leaders) {
+                 return ss::parallel_for_each(
+                   boost::irange<ss::shard_id>(0, ss::smp::count),
+                   [this, &leaders](ss::shard_id shard) {
+                       return ss::smp::submit_to(shard, [this, &leaders] {
+                           return ssx::async_for_each(
+                             leaders,
+                             [this](const ntp_leader_revision& leader) {
+                                 _leaders.local().update_partition_leader(
+                                   leader.ntp,
+                                   leader.revision,
+                                   leader.term,
+                                   leader.leader_id);
+                             });
+                       });
+                   });
+             })
+      .then([] { return update_leadership_reply{}; });
 }
 
 namespace {

src/v/cluster/metadata_dissemination_handler.h

@@ -47,7 +47,7 @@ class metadata_dissemination_handler
 
 private:
     ss::future<update_leadership_reply>
-      do_update_leadership(fragmented_vector<ntp_leader_revision>);
+      do_update_leadership(chunked_vector<ntp_leader_revision>);
 
     ss::sharded<partition_leaders_table>& _leaders;
 }; // namespace cluster

src/v/cluster/metadata_dissemination_service.cc

@@ -395,7 +395,7 @@ ss::future<> metadata_dissemination_service::update_leaders_with_health_report(
 ss::future<> metadata_dissemination_service::dispatch_one_update(
   model::node_id target_id, update_retry_meta& meta) {
     // copy updates to make retries possible
-    fragmented_vector<ntp_leader_revision> updates;
+    chunked_vector<ntp_leader_revision> updates;
     updates.reserve(meta.updates.size());
     std::copy(
       meta.updates.begin(), meta.updates.end(), std::back_inserter(updates));

src/v/cluster/metadata_dissemination_types.h

@@ -104,7 +104,7 @@ struct update_leadership_request_v2
       serde::compat_version<0>> {
     using rpc_adl_exempt = std::true_type;
     static constexpr int8_t version = 0;
-    fragmented_vector<ntp_leader_revision> leaders;
+    chunked_vector<ntp_leader_revision> leaders;
 
     update_leadership_request_v2() noexcept = default;
 
@@ -119,7 +119,7 @@ struct update_leadership_request_v2
     };
 
     update_leadership_request_v2 copy() const {
-        fragmented_vector<ntp_leader_revision> leaders_cp;
+        chunked_vector<ntp_leader_revision> leaders_cp;
         leaders_cp.reserve(leaders.size());
         std::copy(
           leaders.begin(), leaders.end(), std::back_inserter(leaders_cp));
@@ -134,7 +134,7 @@ struct update_leadership_request_v2
     }
 
     explicit update_leadership_request_v2(
-      fragmented_vector<ntp_leader_revision> leaders)
+      chunked_vector<ntp_leader_revision> leaders)
       : leaders(std::move(leaders)) {}
 
     auto serde_fields() { return std::tie(leaders); }

src/v/cluster/tests/serialization_rt_test.cc

@@ -793,7 +793,7 @@ SEASTAR_THREAD_TEST_CASE(serde_reflection_roundtrip) {
       tests::random_named_int<model::term_id>(),
       tests::random_named_int<model::node_id>(),
       tests::random_named_int<model::revision_id>()));
-    fragmented_vector<cluster::ntp_leader_revision> l_revs;
+    chunked_vector<cluster::ntp_leader_revision> l_revs;
     l_revs.emplace_back(
       model::random_ntp(),
       tests::random_named_int<model::term_id>(),

src/v/compat/metadata_dissemination_generator.h

@@ -25,7 +25,7 @@ namespace compat {
 template<>
 struct instance_generator<cluster::update_leadership_request_v2> {
     static cluster::update_leadership_request_v2 random() {
-        fragmented_vector<cluster::ntp_leader_revision> values;
+        chunked_vector<cluster::ntp_leader_revision> values;
         values.emplace_back(
           model::random_ntp(),
           tests::random_named_int<model::term_id>(),