Added migration of tx manager coordinator in recovery mode

src/v/cluster/CMakeLists.txt

@@ -76,6 +76,13 @@ rpcgen(
   INCLUDES ${CMAKE_BINARY_DIR}/src/v
 )
 
+rpcgen(
+  TARGET tx_manager_migrator_rpc
+  IN_FILE ${CMAKE_CURRENT_SOURCE_DIR}/migrations/tx_manager_migrator.json
+  OUT_FILE ${CMAKE_CURRENT_BINARY_DIR}/tx_manager_migrator_service.h
+  INCLUDES ${CMAKE_BINARY_DIR}/src/v
+)
+
 v_cc_library(
   NAME cluster
   SRCS
@@ -194,6 +201,7 @@ v_cc_library(
     cloud_metadata/manifest_downloads.cc
     cloud_metadata/producer_id_recovery_manager.cc
     cloud_metadata/uploader.cc
+    migrations/tx_manager_migrator.cc
   DEPS
     Seastar::seastar
     bootstrap_rpc
@@ -207,6 +215,7 @@ v_cc_library(
     offsets_recovery_rpc
     self_test_rpc
     topic_recovery_status_rpc
+    tx_manager_migrator_rpc
     v::raft
     Roaring::roaring
     absl::flat_hash_map

src/v/cluster/cluster_utils.h

@@ -33,7 +33,7 @@ namespace detail {
 
 template<typename T, typename Fn>
 std::vector<cluster::topic_result>
-create_topic_results(const std::vector<T>& topics, Fn fn) {
+make_error_topic_results(const std::vector<T>& topics, Fn fn) {
     std::vector<cluster::topic_result> results;
     results.reserve(topics.size());
     std::transform(
@@ -61,24 +61,24 @@ requires requires(const T& req) {
     { req.tp_ns } -> std::convertible_to<const model::topic_namespace&>;
 }
 std::vector<topic_result>
-create_topic_results(const std::vector<T>& requests, errc error_code) {
-    return detail::create_topic_results(requests, [error_code](const T& r) {
+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> create_topic_results(
+inline std::vector<topic_result> make_error_topic_results(
   const std::vector<model::topic_namespace>& topics, errc error_code) {
-    return detail::create_topic_results(
+    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> create_topic_results(
+inline std::vector<topic_result> make_error_topic_results(
   const std::vector<custom_assignable_topic_configuration>& requests,
   errc error_code) {
-    return detail::create_topic_results(
+    return detail::make_error_topic_results(
       requests, [error_code](const custom_assignable_topic_configuration& r) {
           return topic_result(r.cfg.tp_ns, error_code);
       });

src/v/cluster/controller.json

@@ -155,6 +155,11 @@
             "name": "remove_plugin",
             "input_type": "remove_plugin_request",
             "output_type": "remove_plugin_response"
+        },
+        {
+            "name": "delete_topics",
+            "input_type": "delete_topics_request",
+            "output_type": "delete_topics_reply"
         }
     ]
 }

src/v/cluster/fwd.h

@@ -70,6 +70,7 @@ class topic_recovery_status_frontend;
 class node_isolation_watcher;
 struct controller_snapshot;
 struct controller_join_snapshot;
+class tx_manager_migrator;
 
 namespace node {
 class local_monitor;