Merge pull request #15049 from mmaslankaprv/fix-14413

Stop recovery when follower offset was already updated

src/v/raft/consensus.cc

@@ -1158,10 +1158,16 @@ ss::future<std::error_code> consensus::replace_configuration(
   model::revision_id new_revision,
   std::optional<model::offset> learner_start_offset) {
     return change_configuration(
-      [nodes = std::move(nodes), new_revision, learner_start_offset](
+      [this, nodes = std::move(nodes), new_revision, learner_start_offset](
         group_configuration current) mutable {
+          auto old = current;
           current.set_version(raft::group_configuration::v_5);
           current.replace(nodes, new_revision, learner_start_offset);
+          vlog(
+            _ctxlog.debug,
+            "Replacing current configuration: {} with new configuration: {}",
+            old,
+            current);
 
           return result<group_configuration>{std::move(current)};
       });

src/v/raft/recovery_stm.cc

@@ -557,6 +557,18 @@ ss::future<> recovery_stm::replicate(
         _stop_requested = true;
         return ss::now();
     }
+    if (meta.value()->expected_log_end_offset >= _last_batch_offset) {
+        vlog(
+          _ctxlog.trace,
+          "follower expected log end offset is already updated, stopping "
+          "recovery. Expected log end offset: {}, recovery range last offset: "
+          "{}",
+          meta.value()->expected_log_end_offset,
+          _last_batch_offset);
+
+        _stop_requested = true;
+        return ss::now();
+    }
     /**
      * Update follower expected log end. It is equal to the last batch in a set
      * of batches read for this recovery round.