cluster: Avoid oversize allocs for topic creation and configuration

src/v/cloud_storage/tests/topic_recovery_service_test.cc

@@ -278,13 +278,13 @@ FIXTURE_TEST(recovery_with_missing_topic_manifest, fixture) {
 }
 
 FIXTURE_TEST(recovery_with_existing_topic, fixture) {
- cluster::topic_configuration cfg{
- model::ns{"kafka"}, model::topic{"test"}, 1, 1};
- std::vector<cluster::custom_assignable_topic_configuration> topic_cfg = {
- cluster::custom_assignable_topic_configuration{std::move(cfg)}};
+ cluster::custom_assignable_topic_configuration_vector topic_cfg{
+ {cluster::custom_assignable_topic_configuration{
+ {model::ns{"kafka"}, model::topic{"test"}, 1, 1}}}};
  auto topic_create_result = app.controller->get_topics_frontend()
  .local()
- .create_topics(topic_cfg, model::no_timeout)
+ .create_topics(
+ std::move(topic_cfg), model::no_timeout)
  .get();
  wait_for_topics(std::move(topic_create_result)).get();
  set_expectations_and_listen(

src/v/cluster/cloud_metadata/cluster_recovery_backend.cc

@@ -230,7 +230,7 @@ ss::future<cluster::errc> cluster_recovery_backend::do_action(
  case recovery_stage::recovered_remote_topic_data: {
  retry_chain_node topics_retry(&parent_retry);
  // TODO: batch this up.
- std::vector<topic_configuration> topics;
+ topic_configuration_vector topics;
  for (size_t i = 0; i < actions.remote_topics.size(); i++) {
  auto& topic_cfg = actions.remote_topics[i];
  if (topic_cfg.is_internal()) {
@@ -259,7 +259,7 @@ ss::future<cluster::errc> cluster_recovery_backend::do_action(
  case recovery_stage::recovered_topic_data: {
  retry_chain_node topics_retry(&parent_retry);
  // TODO: batch this up.
- std::vector<topic_configuration> topics;
+ topic_configuration_vector topics;
  for (size_t i = 0; i < actions.local_topics.size(); i++) {
  topics.emplace_back(std::move(actions.local_topics[i]));
  vlog(clusterlog.debug, "Creating topic {}", topics.back().tp_ns);

src/v/cluster/cluster_utils.cc

@@ -25,6 +25,7 @@
 #include <seastar/core/future.hh>
 
 #include <chrono>
+#include <iterator>
 
 namespace cluster {
 
@@ -110,13 +111,13 @@ bool are_replica_sets_equal(
  return l_sorted == r_sorted;
 }
 
-std::vector<custom_assignable_topic_configuration>
-without_custom_assignments(std::vector<topic_configuration> topics) {
- std::vector<custom_assignable_topic_configuration> assignable_topics;
+custom_assignable_topic_configuration_vector
+without_custom_assignments(topic_configuration_vector topics) {
+ custom_assignable_topic_configuration_vector assignable_topics;
  assignable_topics.reserve(topics.size());
  std::transform(
- topics.begin(),
- topics.end(),
+ std::make_move_iterator(topics.begin()),
+ std::make_move_iterator(topics.end()),
  std::back_inserter(assignable_topics),
  [](topic_configuration cfg) {
  return custom_assignable_topic_configuration(std::move(cfg));

src/v/cluster/cluster_utils.h

@@ -26,24 +26,48 @@
 
 #include <seastar/core/sharded.hh>
 
+#include <concepts>
+#include <ranges>
 #include <system_error>
 #include <utility>
 
 namespace detail {
 
-template<typename T, typename Fn>
+template<std::ranges::range Rng, typename Fn>
+requires std::same_as<
+ std::invoke_result_t<Fn, std::ranges::range_value_t<Rng>>,
+ cluster::topic_result>
 std::vector<cluster::topic_result>
-make_error_topic_results(const std::vector<T>& topics, Fn fn) {
+make_error_topic_results(const Rng& topics, Fn fn) {
  std::vector<cluster::topic_result> results;
  results.reserve(topics.size());
  std::transform(
- topics.cbegin(),
- topics.cend(),
- std::back_inserter(results),
- [&fn](const T& t) { return fn(t); });
+ topics.cbegin(), topics.cend(), std::back_inserter(results), fn);
  return results;
 }
 
+template<typename T>
+concept has_tp_ns = requires {
+ {
+ std::declval<T>().tp_ns
+ } -> std::convertible_to<const model::topic_namespace&>;
+};
+
+template<typename T>
+const model::topic_namespace& extract_tp_ns(const T& t) {
+ if constexpr (std::same_as<T, model::topic_namespace>) {
+ return t;
+ } else if constexpr (has_tp_ns<T>) {
+ return t.tp_ns;
+ } else if constexpr (std::same_as<
+ T,
+ cluster::custom_assignable_topic_configuration>) {
+ return t.cfg.tp_ns;
+ } else {
+ static_assert(always_false_v<T>, "couldn't extract tp_ns");
+ }
+}
+
 } // namespace detail
 
 namespace config {
@@ -56,31 +80,11 @@ class metadata_cache;
 class partition;
 
 /// Creates the same topic_result for all requests
-template<typename T>
-requires requires(const T& req) {
- { req.tp_ns } -> std::convertible_to<const model::topic_namespace&>;
-}
-std::vector<topic_result>
-make_error_topic_results(const std::vector<T>& requests, errc error_code) {
- return detail::make_error_topic_results(requests, [error_code](const T& r) {
- return topic_result(r.tp_ns, error_code);
- });
-}
-
-inline std::vector<topic_result> make_error_topic_results(
- const std::vector<model::topic_namespace>& topics, errc error_code) {
- return detail::make_error_topic_results(
- topics, [error_code](const model::topic_namespace& t) {
- return topic_result(t, error_code);
- });
-}
-
-inline std::vector<topic_result> make_error_topic_results(
- const std::vector<custom_assignable_topic_configuration>& requests,
- errc error_code) {
+std::vector<topic_result> make_error_topic_results(
+ const std::ranges::range auto& topics, errc error_code) {
  return detail::make_error_topic_results(
- requests, [error_code](const custom_assignable_topic_configuration& r) {
- return topic_result(r.cfg.tp_ns, error_code);
+ topics, [error_code](const auto& t) {
+ return topic_result(detail::extract_tp_ns(t), error_code);
  });
 }
 
@@ -189,8 +193,8 @@ ss::future<std::error_code> replicate_and_wait(
  });
 }
 
-std::vector<custom_assignable_topic_configuration>
- without_custom_assignments(std::vector<topic_configuration>);
+custom_assignable_topic_configuration_vector
+ without_custom_assignments(topic_configuration_vector);
 
 template<class T>
 inline std::vector<T>

src/v/cluster/service.cc

@@ -158,10 +158,10 @@ service::purged_topic(purged_topic_request r, rpc::streaming_context&) {
  [](topic_result res) { return purged_topic_reply(std::move(res)); });
 }
 
-std::pair<std::vector<model::topic_metadata>, std::vector<topic_configuration>>
+std::pair<std::vector<model::topic_metadata>, topic_configuration_vector>
 service::fetch_metadata_and_cfg(const std::vector<topic_result>& res) {
  std::vector<model::topic_metadata> md;
- std::vector<topic_configuration> cfg;
+ topic_configuration_vector cfg;
  md.reserve(res.size());
  for (const auto& r : res) {
  if (r.ec == errc::success) {
@@ -237,7 +237,7 @@ service::do_update_topic_properties(update_topic_properties_request req) {
  // local topic frontend instance will eventually dispatch request to _raft0
  // core
  auto res = co_await _topics_frontend.local().update_topic_properties(
- req.updates,
+ std::move(req).updates,
  config::shard_local_cfg().replicate_append_timeout_ms()
  + model::timeout_clock::now());
 

src/v/cluster/service.h

@@ -137,9 +137,8 @@ class service : public controller_service {
 
 private:
  static constexpr auto default_move_interruption_timeout = 10s;
- std::
- pair<std::vector<model::topic_metadata>, std::vector<topic_configuration>>
- fetch_metadata_and_cfg(const std::vector<topic_result>&);
+ std::pair<std::vector<model::topic_metadata>, topic_configuration_vector>
+ fetch_metadata_and_cfg(const std::vector<topic_result>&);
 
  ss::future<finish_partition_update_reply>
  do_finish_partition_update(finish_partition_update_request);

src/v/cluster/tests/autocreate_test.cc

@@ -25,9 +25,9 @@
 #include <chrono>
 #include <vector>
 
-std::vector<cluster::topic_configuration> test_topics_configuration(
+cluster::topic_configuration_vector test_topics_configuration(
  cluster::replication_factor rf = cluster::replication_factor{1}) {
- return std::vector<cluster::topic_configuration>{
+ return cluster::topic_configuration_vector{
  cluster::topic_configuration(test_ns, model::topic("tp-1"), 10, rf),
  cluster::topic_configuration(test_ns, model::topic("tp-2"), 10, rf),
  cluster::topic_configuration(test_ns, model::topic("tp-3"), 10, rf),

src/v/cluster/tests/cluster_test_fixture.h

@@ -221,7 +221,7 @@ class cluster_test_fixture {
  return it.second->app.controller->is_raft0_leader();
  });
  auto& app_0 = leader_it->second->app;
- std::vector<cluster::topic_configuration> cfgs = {
+ cluster::topic_configuration_vector cfgs = {
  cluster::topic_configuration{
  tp_ns.ns, tp_ns.tp, partitions, replication_factor}};
  auto results = app_0.controller->get_topics_frontend()

src/v/cluster/tests/controller_api_tests.cc

@@ -45,13 +45,13 @@ FIXTURE_TEST(test_querying_ntp_status, cluster_test_fixture) {
  auto leader = get_node_application(*leader_id);
 
  // create topic
- std::vector<cluster::topic_configuration> topics;
+ cluster::topic_configuration_vector topics;
  topics.emplace_back(test_ntp.ns, test_ntp.tp.topic, 3, 1);
 
  leader->controller->get_topics_frontend()
  .local()
  .create_topics(
- cluster::without_custom_assignments(topics),
+ cluster::without_custom_assignments(std::move(topics)),
  1s + model::timeout_clock::now())
  .get();
 

src/v/cluster/tests/health_monitor_test.cc

@@ -221,12 +221,11 @@ FIXTURE_TEST(test_ntp_filter, cluster_test_fixture) {
  }).get();
 
  // create topics
- std::vector<cluster::topic_configuration> topics;
- topics.push_back(topic_cfg(model::kafka_namespace, "tp-1", 3, 3));
- topics.push_back(topic_cfg(model::kafka_namespace, "tp-2", 3, 2));
- topics.push_back(topic_cfg(model::kafka_namespace, "tp-3", 3, 1));
- topics.push_back(
- topic_cfg(model::kafka_internal_namespace, "internal-1", 3, 2));
+ cluster::topic_configuration_vector topics{
+ topic_cfg(model::kafka_namespace, "tp-1", 3, 3),
+ topic_cfg(model::kafka_namespace, "tp-2", 3, 2),
+ topic_cfg(model::kafka_namespace, "tp-3", 3, 1),
+ topic_cfg(model::kafka_internal_namespace, "internal-1", 3, 2)};
 
  n1->controller->get_topics_frontend()
  .local()

src/v/cluster/tests/metadata_dissemination_test.cc

@@ -88,7 +88,7 @@ FIXTURE_TEST(
  BOOST_REQUIRE_EQUAL(cache_2.node_count(), 3);
 
  // Create topic with replication factor 1
- std::vector<cluster::topic_configuration> topics;
+ cluster::topic_configuration_vector topics;
  topics.emplace_back(model::ns("default"), model::topic("test_1"), 3, 1);
  cntrl_0->controller->get_topics_frontend()
  .local()
@@ -122,7 +122,7 @@ FIXTURE_TEST(test_metadata_dissemination_joining_node, cluster_test_fixture) {
  BOOST_REQUIRE_EQUAL(cache_1.node_count(), 2);
 
  // Create topic with replication factor 1
- std::vector<cluster::topic_configuration> topics;
+ cluster::topic_configuration_vector topics;
  topics.emplace_back(model::ns("default"), model::topic("test_1"), 3, 1);
  cntrl_0->controller->get_topics_frontend()
  .local()

src/v/cluster/tests/serialization_rt_test.cc

@@ -1337,7 +1337,7 @@ SEASTAR_THREAD_TEST_CASE(serde_reflection_roundtrip) {
  roundtrip_test(data);
  }
  {
- std::vector<cluster::topic_properties_update> updates;
+ cluster::topic_properties_update_vector updates;
  for (int i = 0, mi = random_generators::get_int(10); i < mi; i++) {
  cluster::property_update<std::optional<v8_engine::data_policy>>
  data_policy;
@@ -1350,16 +1350,15 @@ SEASTAR_THREAD_TEST_CASE(serde_reflection_roundtrip) {
  cluster::incremental_topic_custom_updates custom_properties{
  .data_policy = data_policy,
  };
- updates.push_back(cluster::topic_properties_update{
+ updates.emplace_back(
  model::random_topic_namespace(),
  random_incremental_topic_updates(),
- custom_properties,
- });
+ custom_properties);
  }
  cluster::update_topic_properties_request data{
- .updates = updates,
+ .updates = std::move(updates),
  };
- roundtrip_test(data);
+ roundtrip_test(std::move(data));
  }
  {
  cluster::topic_result data{
@@ -1698,18 +1697,18 @@ SEASTAR_THREAD_TEST_CASE(serde_reflection_roundtrip) {
  roundtrip_test(data);
  }
  {
- std::vector<cluster::topic_configuration> topics;
+ cluster::topic_configuration_vector topics;
  for (auto i = 0, mi = random_generators::get_int(20); i < mi; ++i) {
  topics.push_back(random_topic_configuration());
  }
  cluster::create_topics_request data{
- .topics = topics,
+ .topics = std::move(topics),
  .timeout = random_timeout_clock_duration(),
  };
  // adl encoding for topic_configuration doesn't encode/decode to exact
  // equality, but also already existed prior to serde support being added
  // so only testing the serde case.
- roundtrip_test(data);
+ roundtrip_test(std::move(data));
  }
  {
  auto data = random_partition_metadata();
@@ -1743,7 +1742,7 @@ SEASTAR_THREAD_TEST_CASE(serde_reflection_roundtrip) {
  // adl serialization doesn't preserve equality for topic_configuration.
  // serde serialization does and was added after support for adl so adl
  // semantics are preserved.
- roundtrip_test(data);
+ roundtrip_test(std::move(data));
  }
  {
  raft::transfer_leadership_request data{

src/v/cluster/topic_recovery_service.cc

@@ -16,6 +16,7 @@
 #include "cloud_storage/topic_manifest.h"
 #include "cluster/topic_recovery_status_frontend.h"
 #include "cluster/topics_frontend.h"
+#include "cluster/types.h"
 
 #include <seastar/http/request.hh>
 #include <seastar/util/defer.hh>
@@ -410,7 +411,7 @@ static cluster::topic_configuration make_topic_config(
 
 ss::future<std::vector<cluster::topic_result>>
 topic_recovery_service::create_topics(const recovery_request& request) {
- std::vector<cluster::topic_configuration> topic_configs;
+ cluster::topic_configuration_vector topic_configs;
  topic_configs.reserve(_downloaded_manifests->size());
 
  std::transform(
@@ -420,7 +421,8 @@ topic_recovery_service::create_topics(const recovery_request& request) {
  [&request](const auto& m) { return make_topic_config(m, request); });
 
  co_return co_await _topics_frontend.local().autocreate_topics(
- topic_configs, config::shard_local_cfg().create_topic_timeout_ms());
+ std::move(topic_configs),
+ config::shard_local_cfg().create_topic_timeout_ms());
 }
 
 ss::future<std::vector<cloud_storage::topic_manifest>>
@@ -532,7 +534,7 @@ ss::future<> topic_recovery_service::reset_topic_configurations() {
  co_return;
  }
 
- std::vector<cluster::topic_properties_update> updates;
+ cluster::topic_properties_update_vector updates;
  updates.reserve(_downloaded_manifests->size());
  std::transform(
  std::make_move_iterator(_downloaded_manifests->begin()),