controller/api: introduce get_partition_leader_reconciliation_state

Author: bharathv

src/v/cluster/controller_api.cc

@@ -20,6 +20,7 @@
 #include "cluster/partition_balancer_backend.h"
 #include "cluster/partition_balancer_rpc_service.h"
 #include "cluster/partition_balancer_types.h"
+#include "cluster/partition_leaders_table.h"
 #include "cluster/partition_manager.h"
 #include "cluster/shard_table.h"
 #include "cluster/topic_table.h"
@@ -285,6 +286,22 @@ controller_api::get_reconciliation_state(
       });
 }
 
+ss::future<result<ntp_reconciliation_state>>
+controller_api::get_partition_leader_reconciliation_state(
+  model::ntp ntp, model::timeout_clock::time_point timeout) {
+    auto leader = _partition_leaders.local().get_leader(ntp);
+    if (!leader) {
+        vlog(
+          clusterlog.debug,
+          "can't get partition leader for ntp {} to get its reconciliation "
+          "state",
+          ntp);
+        co_return result<ntp_reconciliation_state>(errc::no_leader_controller);
+    }
+    co_return co_await get_reconciliation_state(
+      *leader, std::move(ntp), timeout);
+}
+
 ss::future<result<ntp_reconciliation_state>>
 controller_api::get_reconciliation_state(
   model::node_id id, model::ntp ntp, model::timeout_clock::time_point timeout) {

src/v/cluster/controller_api.h

@@ -66,6 +66,10 @@ class controller_api {
     ss::future<result<ntp_reconciliation_state>> get_reconciliation_state(
       model::node_id, model::ntp, model::timeout_clock::time_point);
 
+    ss::future<result<ntp_reconciliation_state>>
+      get_partition_leader_reconciliation_state(
+        model::ntp, model::timeout_clock::time_point);
+
     // high level APIs
     ss::future<std::error_code> wait_for_topic(
       model::topic_namespace_view, model::timeout_clock::time_point);
@@ -107,7 +111,7 @@ class controller_api {
     ss::sharded<members_table>& _members;
     ss::sharded<partition_balancer_backend>& _partition_balancer;
     ss::sharded<partition_manager>& _partition_manager;
-    [[maybe_unused]] ss::sharded<partition_leaders_table>& _partition_leaders;
+    ss::sharded<partition_leaders_table>& _partition_leaders;
     ss::sharded<ss::abort_source>& _as;
 };
 } // namespace cluster