[ballot] Update ballot init function signature

src/braft/ballot.cpp

@@ -21,7 +21,7 @@ namespace braft {
 Ballot::Ballot() : _quorum(0), _old_quorum(0) {}
 Ballot::~Ballot() {}
 
-int Ballot::init(const Configuration& conf, const Configuration* old_conf) {
+void Ballot::init(const Configuration& conf, const Configuration* old_conf) {
     _peers.clear();
     _old_peers.clear();
     _quorum = 0;
@@ -34,15 +34,14 @@ int Ballot::init(const Configuration& conf, const Configuration* old_conf) {
     }
     _quorum = _peers.size() / 2 + 1;
     if (!old_conf) {
-        return 0;
+        return;
     }
     _old_peers.reserve(old_conf->size());
     for (Configuration::const_iterator
             iter = old_conf->begin(); iter != old_conf->end(); ++iter) {
         _old_peers.push_back(*iter);
     }
     _old_quorum = _old_peers.size() / 2 + 1;
-    return 0;
 }
 
 Ballot::PosHint Ballot::grant(const PeerId& peer, PosHint hint) {

src/braft/ballot.h

@@ -38,7 +38,7 @@ class Ballot {
         std::swap(_old_quorum, rhs._old_quorum);
     }
 
-    int init(const Configuration& conf, const Configuration* old_conf);
+    void init(const Configuration& conf, const Configuration* old_conf);
     PosHint grant(const PeerId& peer, PosHint hint);
     void grant(const PeerId& peer);
     bool granted() const { return _quorum <= 0 && _old_quorum <= 0; }

src/braft/ballot_box.cpp

@@ -121,10 +121,7 @@ int BallotBox::reset_pending_index(int64_t new_pending_index) {
 int BallotBox::append_pending_task(const Configuration& conf, const Configuration* old_conf,
                                    Closure* closure) {
     Ballot bl;
-    if (bl.init(conf, old_conf) != 0) {
-        CHECK(false) << "Fail to init ballot";
-        return -1;
-    }
+    bl.init(conf, old_conf);
 
     BAIDU_SCOPED_LOCK(_mutex);
     CHECK(_pending_index > 0);

src/braft/replicator.cpp

@@ -447,15 +447,23 @@ void Replicator::_on_rpc_returned(ReplicatorId id, brpc::Controller* cntl,
                        << " is " << response->last_log_index();
             // The peer contains less logs than leader
             r->_next_index = response->last_log_index() + 1;
-        } else {  
+        } else {
             // The peer contains logs from old term which should be truncated,
             // decrease _last_log_at_peer by one to test the right index to keep
             if (BAIDU_LIKELY(r->_next_index > 1)) {
-                BRAFT_VLOG << "Group " << r->_options.group_id 
+                BRAFT_VLOG << "Group " << r->_options.group_id
                            << " log_index=" << r->_next_index << " mismatch";
-                --r->_next_index;
+                // Decrease the |_next_index| when the request with the minimum log
+                // index mismatch.
+                // Because when we enable pipeline replication and disable cache,
+                // the request with larger log index would be handled before the
+                // smaller request, we should ignore it.
+                // See https://github.com/baidu/braft/issues/421
+                if (request->prev_log_index() == r->_next_index - 1) {
+                    --r->_next_index;
+                }
             } else {
-                LOG(ERROR) << "Group " << r->_options.group_id 
+                LOG(ERROR) << "Group " << r->_options.group_id
                            << " peer=" << r->_options.peer_id
                            << " declares that log at index=0 doesn't match,"
                               " which is not supposed to happen";

test/test_ballot.cpp

@@ -29,7 +29,7 @@ TEST(BallotTest, sanity) {
     conf.add_peer(peer2);
     conf.add_peer(peer3);
     braft::Ballot bl;
-    ASSERT_EQ(0, bl.init(conf, NULL));
+    bl.init(conf, NULL);
     ASSERT_EQ(2, bl._quorum);
     ASSERT_EQ(0, bl._old_quorum);
     bl.grant(peer1);
@@ -54,7 +54,7 @@ TEST(BallotTest, joint_consensus_same_conf) {
     conf.add_peer(peer2);
     conf.add_peer(peer3);
     braft::Ballot bl;
-    ASSERT_EQ(0, bl.init(conf, &conf));
+    bl.init(conf, &conf);
     ASSERT_EQ(2, bl._quorum);
     ASSERT_EQ(2, bl._old_quorum);
     bl.grant(peer1);
@@ -92,7 +92,7 @@ TEST(BallotTest, joint_consensus_different_conf) {
     conf2.add_peer(peer3);
     conf2.add_peer(peer4);
     braft::Ballot bl;
-    ASSERT_EQ(0, bl.init(conf, &conf2));
+    bl.init(conf, &conf2);
     bl.grant(peer1);
     bl.grant(peer2);
     ASSERT_FALSE(bl.granted());