Implement on_rollback()

Signed-off-by: Xiaoxi Chen <xiaoxchen@ebay.com>

Author: xiaoxichen

src/lib/replication/repl_dev/raft_state_machine.cpp

@@ -179,6 +179,7 @@ raft_buf_ptr_t RaftStateMachine::pre_commit_ext(nuraft::state_machine::ext_op_pa
 
     repl_req_ptr_t rreq = lsn_to_req(lsn);
     RD_LOGD("Raft channel: Precommit rreq=[{}]", rreq->to_compact_string());
+    // Fixme: Check return value of on_pre_commit
     m_rd.m_listener->on_pre_commit(rreq->lsn(), rreq->header(), rreq->key(), rreq);
 
     return m_success_ptr;
@@ -195,12 +196,23 @@ raft_buf_ptr_t RaftStateMachine::commit_ext(nuraft::state_machine::ext_op_params
         // This is the time to ensure flushing of journal happens in the proposer
         rreq->add_state(repl_req_state_t::LOG_FLUSHED);
     }
-
+    // Fixme: Check return value of handle_commit
     m_rd.handle_commit(rreq);
 
     return m_success_ptr;
 }
 
+void RaftStateMachine::rollback_ext(const nuraft::state_machine::ext_op_params& params) {
+    int64_t lsn = s_cast< int64_t >(params.log_idx);
+    RD_LOGD("Raft channel: Received rollback message lsn {}, store {}, logdev {}", lsn,
+            m_rd.m_data_journal->logstore_id(), m_rd.m_data_journal->logdev_id());
+    repl_req_ptr_t rreq = lsn_to_req(lsn);
+    RD_DBG_ASSERT(rreq != nullptr, "Raft channel got null rreq");
+    RD_LOGD("Raft channel: rollback rreq=[{}]", rreq->to_compact_string());
+    // Fixme: Check return value of on_rollback
+    m_rd.m_listener->on_rollback(rreq->lsn(), rreq->header(), rreq->key(), rreq);
+}
+
 void RaftStateMachine::iterate_repl_reqs(std::function< void(int64_t, repl_req_ptr_t rreq) > const& cb) {
     for (auto [key, rreq] : m_lsn_req_map) {
         cb(key, rreq);

src/lib/replication/repl_dev/raft_state_machine.h

@@ -108,7 +108,8 @@ class RaftStateMachine : public nuraft::state_machine {
     uint64_t last_commit_index() override;
     raft_buf_ptr_t pre_commit_ext(const nuraft::state_machine::ext_op_params& params) override;
     raft_buf_ptr_t commit_ext(const nuraft::state_machine::ext_op_params& params) override;
-    void rollback(uint64_t lsn, nuraft::buffer&) override { LOGCRITICAL("Unimplemented rollback on: [{}]", lsn); }
+    void rollback_ext(const nuraft::state_machine::ext_op_params& params) override;
+    // void rollback(uint64_t lsn, nuraft::buffer&) override { LOGCRITICAL("Unimplemented rollback on: [{}]", lsn); }
     void become_ready();
 
     bool apply_snapshot(nuraft::snapshot&) override { return false; }