reset admin

Author: andrwng

src/v/redpanda/admin/services/datalake/datalake.cc

@@ -116,9 +116,33 @@ datalake_service_impl::datalake_service_impl(
   : _proxy_client(std::move(proxy_client))
   , _coordinator_fe(coordinator_fe) {}
 
+<<<<<<< HEAD
 ss::future<proto::admin::get_coordinator_state_response>
 datalake_service_impl::get_coordinator_state(
   serde::pb::rpc::context, proto::admin::get_coordinator_state_request req) {
+=======
+ss::future<proto::admin::coordinator_reset_pending_state_response>
+datalake_service_impl::coordinator_reset_pending_state(
+  serde::pb::rpc::context,
+  proto::admin::coordinator_reset_pending_state_request req) {
+    datalake::coordinator::reset_pending_state_request fe_req;
+    fe_req.topic = model::topic{req.get_topic_name()};
+    if (!_coordinator_fe->local_is_initialized()) {
+        throw serde::pb::rpc::unavailable_exception(
+          "Datalake coordinator frontend not initialized");
+    }
+    auto fe_res = co_await _coordinator_fe->local().reset_pending_state(fe_req);
+    if (fe_res.errc != datalake::coordinator::errc::ok) {
+        throw serde::pb::rpc::internal_exception(
+          fmt::format("Datalake coordinator error: {}", fe_res.errc));
+    }
+    co_return proto::admin::coordinator_reset_pending_state_response{};
+}
+
+ss::future<proto::admin::coordinator_get_state_response>
+datalake_service_impl::coordinator_get_state(
+  serde::pb::rpc::context, proto::admin::coordinator_get_state_request req) {
+>>>>>>> 120d3b7539 (reset admin)
     if (!_coordinator_fe->local_is_initialized()) {
         throw serde::pb::rpc::unavailable_exception(
           "Datalake coordinator frontend not initialized");

src/v/redpanda/admin/services/datalake/datalake.h

@@ -28,6 +28,11 @@ class datalake_service_impl : public proto::admin::datalake_service {
         serde::pb::rpc::context,
         proto::admin::get_coordinator_state_request) override;
 
+    ss::future<proto::admin::coordinator_reset_pending_state_response>
+      coordinator_reset_pending_state(
+        serde::pb::rpc::context,
+        proto::admin::coordinator_reset_pending_state_request) override;
+
 private:
     admin::proxy::client _proxy_client;