From 8ae0c0d54e0fa4c2c67eaf95bf82a63a6a246236 Mon Sep 17 00:00:00 2001
From: Michal Maslanka <michal@redpanda.com>
Date: Thu, 6 Jul 2023 12:41:25 +0200
Subject: [PATCH] c/backend: wait for leader to be elected before finishing
 abort cancel

When forcibly aborting reconfiguration we should wait for the new leader
to be elected in the configuration that the partition was forced to.
This way we can be certain that the new configuration will finally be
replicated to the majority of nodes even tough the leader may not
exists at the time when configuration is replicated.

Fixes: #9243

Signed-off-by: Michal Maslanka <michal@redpanda.com>
(cherry picked from commit 3948312e2c2af16049b80987344c3822b4c5d0f2)
---
 src/v/cluster/controller_backend.cc | 9 +++++++--
 src/v/cluster/controller_backend.h  | 1 +
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/src/v/cluster/controller_backend.cc b/src/v/cluster/controller_backend.cc
index 73196a94ee1c5..dd4a8f98730ec 100644
--- a/src/v/cluster/controller_backend.cc
+++ b/src/v/cluster/controller_backend.cc
@@ -959,7 +959,10 @@ controller_backend::process_partition_reconfiguration(
              * configuration so old replicas are not longer needed.
              */
             if (can_finish_update(
-                  type, target_assignment.replicas, previous_replicas)) {
+                  partition->get_leader_id(),
+                  type,
+                  target_assignment.replicas,
+                  previous_replicas)) {
                 co_return co_await dispatch_update_finished(
                   std::move(ntp), target_assignment);
             }
@@ -1009,12 +1012,14 @@ controller_backend::process_partition_reconfiguration(
 }
 
 bool controller_backend::can_finish_update(
+  std::optional<model::node_id> current_leader,
   topic_table_delta::op_type update_type,
   const std::vector<model::broker_shard>& current_replicas,
   const std::vector<model::broker_shard>& previous_replicas) {
     // force abort update may be finished by any node
     if (update_type == topic_table_delta::op_type::force_abort_update) {
-        return true;
+        return current_leader == _self;
+        ;
     }
     // update may be finished by a node that was added to replica set
     if (!has_local_replicas(_self, previous_replicas)) {
diff --git a/src/v/cluster/controller_backend.h b/src/v/cluster/controller_backend.h
index 47d5e0f83a5ae..4f9fc093ecd18 100644
--- a/src/v/cluster/controller_backend.h
+++ b/src/v/cluster/controller_backend.h
@@ -340,6 +340,7 @@ class controller_backend
       model::ntp, ss::shard_id, partition_assignment);
 
     bool can_finish_update(
+      std::optional<model::node_id>,
       topic_table_delta::op_type,
       const std::vector<model::broker_shard>&,
       const std::vector<model::broker_shard>&);