Merge pull request #22982 from vbotbuildovich/backport-pr-22977-v24.2…

….x-717

[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>(),