SERVER-39831 Never update commit point beyond last applied if learned from sync source

9 files changed, 302 insertions, 36 deletions

diff --git a/jstests/replsets/do_not_advance_commit_point_beyond_last_applied_term.js b/jstests/replsets/do_not_advance_commit_point_beyond_last_applied_term.js
new file mode 100644
index 00000000000..c57f80d3719
--- /dev/null
+++ b/jstests/replsets/do_not_advance_commit_point_beyond_last_applied_term.js
@@ -0,0 +1,112 @@
+/**
+ * Tests that secondaries do not advance their commit point to have a greater term than their
+ * lastApplied. This prevents incorrectly prefix-committing operations that are not majority
+ * committed.
+ * @tags: [requires_majority_read_concern]
+ */
+(function() {
+    "use strict";
+
+    load("jstests/libs/write_concern_util.js");  // for [stop|restart]ServerReplication.
+
+    const dbName = "test";
+    const collName = "coll";
+
+    // Set up a ReplSetTest where nodes only sync one oplog entry at a time.
+    const rst = new ReplSetTest(
+        {nodes: 5, useBridge: true, nodeOptions: {setParameter: "bgSyncOplogFetcherBatchSize=1"}});
+    rst.startSet();
+    const config = rst.getReplSetConfig();
+    // Prevent elections.
+    config.settings = {electionTimeoutMillis: 12 * 60 * 60 * 1000};
+    rst.initiate(config);
+
+    const nodeA = rst.nodes[0];
+    const nodeB = rst.nodes[1];
+    const nodeC = rst.nodes[2];
+    const nodeD = rst.nodes[3];
+    const nodeE = rst.nodes[4];
+
+    jsTest.log("Node A is primary in term 1. Node E is delayed.");
+    // A: [1]
+    // B: [1]
+    // C: [1]
+    // D: [1]
+    // E:
+    assert.eq(nodeA, rst.getPrimary());
+    nodeE.disconnect([nodeA, nodeB, nodeC, nodeD]);
+    assert.commandWorked(nodeA.getDB(dbName)[collName].insert({term: 1}));
+    rst.awaitReplication(undefined, undefined, [nodeB, nodeC, nodeD]);
+
+    jsTest.log("Node B steps up in term 2 and performs a write, which is not replicated.");
+    // A: [1]
+    // B: [1] [2]
+    // C: [1]
+    // D: [1]
+    // E:
+    stopServerReplication([nodeA, nodeC, nodeD]);
+    assert.commandWorked(nodeB.adminCommand({replSetStepUp: 1}));
+    rst.waitForState(nodeA, ReplSetTest.State.SECONDARY);
+    assert.eq(nodeB, rst.getPrimary());
+    assert.commandWorked(nodeB.getDB(dbName)[collName].insert({term: 2}));
+
+    jsTest.log("Node A steps up again in term 3 with votes from A, C, and D and commits a write.");
+    // A: [1] [3]
+    // B: [1] [2]
+    // C: [1] [3]
+    // D: [1] [3]
+    // E:
+    nodeB.disconnect([nodeA, nodeC, nodeD, nodeE]);
+    assert.commandWorked(nodeA.adminCommand({replSetStepUp: 1}));
+    restartServerReplication([nodeA, nodeC, nodeD]);
+    assert.soon(() => {
+        // We cannot use getPrimary() here because 2 nodes report they are primary.
+        return assert.commandWorked(nodeA.adminCommand({ismaster: 1})).ismaster;
+    });
+    assert.commandWorked(
+        nodeA.getDB(dbName)[collName].insert({term: 3}, {writeConcern: {w: "majority"}}));
+    assert.eq(1, nodeC.getDB(dbName)[collName].find({term: 3}).itcount());
+    assert.eq(1, nodeD.getDB(dbName)[collName].find({term: 3}).itcount());
+
+    jsTest.log("Node E syncs from a majority node and learns the new commit point in term 3.");
+    // A: [1] [3]
+    // B: [1] [2]
+    // C: [1] [3]
+    // D: [1] [3]
+    // E: [1]
+    // The stopReplProducerOnDocument failpoint ensures that Node E stops replicating before
+    // applying the document {msg: "new primary"}, which is the first document of term 3. This
+    // depends on the oplog fetcher batch size being 1.
+    assert.commandWorked(nodeE.adminCommand({
+        configureFailPoint: "stopReplProducerOnDocument",
+        mode: "alwaysOn",
+        data: {document: {msg: "new primary"}}
+    }));
+    nodeE.reconnect([nodeA, nodeC, nodeD]);
+    assert.soon(() => {
+        return 1 === nodeE.getDB(dbName)[collName].find({term: 1}).itcount();
+    });
+    assert.eq(0, nodeE.getDB(dbName)[collName].find({term: 3}).itcount());
+
+    jsTest.log("Node E switches its sync source to B and replicates the stale branch of term 2.");
+    nodeE.disconnect([nodeA, nodeC, nodeD]);
+    nodeB.reconnect(nodeE);
+    assert.commandWorked(
+        nodeE.adminCommand({configureFailPoint: "stopReplProducerOnDocument", mode: "off"}));
+    assert.soon(() => {
+        return 1 === nodeE.getDB(dbName)[collName].find({term: 2}).itcount();
+    });
+
+    jsTest.log("Node E must not return the entry in term 2 as committed.");
+    assert.eq(0, nodeE.getDB(dbName)[collName].find({term: 2}).readConcern("majority").itcount());
+
+    jsTest.log("Reconnect the set. Node E must roll back successfully.");
+    nodeE.reconnect([nodeA, nodeC, nodeD]);
+    nodeB.reconnect([nodeA, nodeC, nodeD]);
+    rst.awaitReplication();
+    assert.eq(1, nodeE.getDB(dbName)[collName].find({term: 1}).itcount());
+    assert.eq(0, nodeE.getDB(dbName)[collName].find({term: 2}).itcount());
+    assert.eq(1, nodeE.getDB(dbName)[collName].find({term: 3}).itcount());
+
+    rst.stopSet();
+}());
diff --git a/jstests/replsets/update_commit_point_from_sync_source_ignores_term.js b/jstests/replsets/update_commit_point_from_sync_source_ignores_term.js
new file mode 100644
index 00000000000..7915dfd4b7b
--- /dev/null
+++ b/jstests/replsets/update_commit_point_from_sync_source_ignores_term.js
@@ -0,0 +1,87 @@
+/**
+ * Tests that even if the sync source's lastOpCommitted is in a higher term than the node's
+ * lastApplied, the node can update its own lastOpCommitted to its lastApplied.
+ * @tags: [requires_majority_read_concern]
+ */
+(function() {
+    "use strict";
+
+    load("jstests/libs/write_concern_util.js");  // for [stop|restart]ServerReplication.
+
+    const dbName = "test";
+    const collName = "coll";
+
+    // Set up a ReplSetTest where nodes only sync one oplog entry at a time.
+    const rst = new ReplSetTest(
+        {nodes: 5, useBridge: true, nodeOptions: {setParameter: "bgSyncOplogFetcherBatchSize=1"}});
+    rst.startSet();
+    const config = rst.getReplSetConfig();
+    // Ban chaining and prevent elections.
+    config.settings = {chainingAllowed: false, electionTimeoutMillis: 12 * 60 * 60 * 1000};
+    rst.initiate(config);
+
+    const nodeA = rst.nodes[0];
+    const nodeB = rst.nodes[1];
+    const nodeC = rst.nodes[2];
+    const nodeD = rst.nodes[3];
+    const nodeE = rst.nodes[4];
+
+    jsTest.log("Node A is primary in term 1. Replicate a write to Node E that is not committed.");
+    assert.eq(nodeA, rst.getPrimary());
+    // Ensure Node E has a majority committed snapshot.
+    assert.commandWorked(nodeA.getDB(dbName)[collName].insert({_id: "dummy"}));
+    rst.awaitLastOpCommitted();
+    stopServerReplication([nodeB, nodeC, nodeD]);
+    assert.commandWorked(nodeA.getDB(dbName)[collName].insert({_id: "term 1, doc 1"}));
+    rst.awaitReplication(undefined, undefined, [nodeE]);
+    assert.eq(0,
+              nodeE.getDB(dbName)[collName]
+                  .find({_id: "term 1, doc 1"})
+                  .readConcern("majority")
+                  .itcount());
+
+    jsTest.log("Disconnect Node E. Perform a new write.");
+    nodeE.disconnect([nodeA, nodeB, nodeC, nodeD]);
+    restartServerReplication([nodeB, nodeC, nodeD]);
+    assert.commandWorked(nodeA.getDB(dbName)[collName].insert({_id: "term 1, doc 2"}));
+
+    jsTest.log("Step up Node B in term 2. Commit a new write.");
+    // Ensure Node B is caught up, so that it can become primary.
+    rst.awaitReplication(undefined, undefined, [nodeB]);
+    assert.commandWorked(nodeB.adminCommand({replSetStepUp: 1}));
+    rst.waitForState(nodeA, ReplSetTest.State.SECONDARY);
+    assert.eq(nodeB, rst.getPrimary());
+    assert.commandWorked(
+        nodeB.getDB(dbName)[collName].insert({_id: "term 2"}, {writeConcern: {w: "majority"}}));
+    // Node E might sync from Node A or Node B. Ensure they both have the new commit point.
+    rst.awaitLastOpCommitted(undefined, [nodeA]);
+
+    jsTest.log("Allow Node E to replicate the last write from term 1.");
+    // The stopReplProducerOnDocument failpoint ensures that Node E stops replicating before
+    // applying the document {msg: "new primary"}, which is the first document of term 2. This
+    // depends on the oplog fetcher batch size being 1.
+    assert.commandWorked(nodeE.adminCommand({
+        configureFailPoint: "stopReplProducerOnDocument",
+        mode: "alwaysOn",
+        data: {document: {msg: "new primary"}}
+    }));
+    nodeE.reconnect([nodeA, nodeB, nodeC, nodeD]);
+    assert.soon(() => {
+        return nodeE.getDB(dbName)[collName].find({_id: "term 1, doc 2"}).itcount() === 1;
+    });
+    assert.eq(0, nodeE.getDB(dbName)[collName].find({_id: "term 2"}).itcount());
+
+    jsTest.log("Node E now knows that its first write is majority committed.");
+    // It does not yet know that {_id: "term 1, doc 2"} is committed. Its last batch was {_id: "term
+    // 1, doc 2"}. The sync source's lastOpCommitted was in term 2, so Node E updated its
+    // lastOpCommitted to its lastApplied, which did not yet include {_id: "term 1, doc 2"}.
+    assert.eq(1,
+              nodeE.getDB(dbName)[collName]
+                  .find({_id: "term 1, doc 1"})
+                  .readConcern("majority")
+                  .itcount());
+
+    assert.commandWorked(
+        nodeE.adminCommand({configureFailPoint: "stopReplProducerOnDocument", mode: "off"}));
+    rst.stopSet();
+}());
diff --git a/src/mongo/db/repl/oplog_fetcher.cpp b/src/mongo/db/repl/oplog_fetcher.cpp
index ce499fd0c89..a9f496d99ad 100644
--- a/src/mongo/db/repl/oplog_fetcher.cpp
+++ b/src/mongo/db/repl/oplog_fetcher.cpp
@@ -51,6 +51,7 @@ namespace repl {
 Seconds OplogFetcher::kDefaultProtocolZeroAwaitDataTimeout(2);
 
 MONGO_FAIL_POINT_DEFINE(stopReplProducer);
+MONGO_FAIL_POINT_DEFINE(stopReplProducerOnDocument);
 
 namespace {
 
@@ -402,6 +403,18 @@ StatusWith<BSONObj> OplogFetcher::_onSuccessfulBatch(const Fetcher::QueryRespons
         return Status(ErrorCodes::FailPointEnabled, "stopReplProducer fail point is enabled");
     }
 
+    // Stop fetching and return when we reach a particular document. This failpoint should be used
+    // with the setParameter bgSyncOplogFetcherBatchSize=1, so that documents are fetched one at a
+    // time.
+    MONGO_FAIL_POINT_BLOCK(stopReplProducerOnDocument, fp) {
+        if (!queryResponse.documents.empty() &&
+            SimpleBSONObjComparator::kInstance.evaluate(
+                fp.getData()["document"].Obj() == queryResponse.documents.front()["o"].Obj())) {
+            return Status(ErrorCodes::FailPointEnabled,
+                          "stopReplProducerOnDocument fail point is enabled");
+        }
+    }
+
     const auto& documents = queryResponse.documents;
     auto firstDocToApply = documents.cbegin();
 
diff --git a/src/mongo/db/repl/replication_coordinator_impl.h b/src/mongo/db/repl/replication_coordinator_impl.h
index bd1d4a26f86..85ac51b26ba 100644
--- a/src/mongo/db/repl/replication_coordinator_impl.h
+++ b/src/mongo/db/repl/replication_coordinator_impl.h
@@ -938,8 +938,8 @@ private:
         stdx::unique_lock<stdx::mutex> lock);
 
     /**
-     * Updates the last committed OpTime to be "committedOpTime" if it is more recent than the
-     * current last committed OpTime.
+     * Updates the last committed OpTime to be min(committedOpTime, lastApplied) if it is more
+     * recent than the current last committed OpTime.
      */
     void _advanceCommitPoint_inlock(const OpTime& committedOpTime);
 
diff --git a/src/mongo/db/repl/replication_coordinator_impl_test.cpp b/src/mongo/db/repl/replication_coordinator_impl_test.cpp
index dc1d39bb82a..8d0b7eac9b9 100644
--- a/src/mongo/db/repl/replication_coordinator_impl_test.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl_test.cpp
@@ -3955,24 +3955,22 @@ TEST_F(StableOpTimeTest, SetMyLastAppliedSetsStableOpTimeForStorage) {
 
     auto repl = getReplCoord();
     Timestamp stableTimestamp;
-    long long term = 2;
 
     getStorageInterface()->supportsDocLockingBool = true;
-
-    repl->advanceCommitPoint(OpTime({1, 1}, term));
     ASSERT_EQUALS(Timestamp::min(), getStorageInterface()->getStableTimestamp());
     ASSERT_OK(getReplCoord()->setFollowerMode(MemberState::RS_SECONDARY));
 
-    getReplCoord()->setMyLastAppliedOpTime(OpTimeWithTermOne(1, 1));
-    getReplCoord()->setMyLastDurableOpTime(OpTimeWithTermOne(1, 1));
+    getStorageInterface()->allCommittedTimestamp = Timestamp(1, 1);
+    repl->setMyLastAppliedOpTime(OpTimeWithTermOne(1, 1));
+    repl->setMyLastDurableOpTime(OpTimeWithTermOne(1, 1));
     simulateSuccessfulV1Election();
 
-    repl->advanceCommitPoint(OpTime({2, 2}, term));
+    // Advance the commit point so it's higher than all the others.
+    repl->advanceCommitPoint(OpTimeWithTermOne(10, 1));
     ASSERT_EQUALS(Timestamp(1, 1), getStorageInterface()->getStableTimestamp());
 
     // Check that the stable timestamp is not updated if the all-committed timestamp is behind.
-    getStorageInterface()->allCommittedTimestamp = Timestamp(1, 1);
-    repl->setMyLastAppliedOpTime(OpTime({1, 2}, term));
+    repl->setMyLastAppliedOpTime(OpTimeWithTermOne(1, 2));
     stableTimestamp = getStorageInterface()->getStableTimestamp();
     ASSERT_EQUALS(Timestamp(1, 1), getStorageInterface()->getStableTimestamp());
 
@@ -3980,17 +3978,17 @@ TEST_F(StableOpTimeTest, SetMyLastAppliedSetsStableOpTimeForStorage) {
 
     // Check that the stable timestamp is updated for the storage engine when we set the applied
     // optime.
-    repl->setMyLastAppliedOpTime(OpTime({2, 1}, term));
+    repl->setMyLastAppliedOpTime(OpTimeWithTermOne(2, 1));
     stableTimestamp = getStorageInterface()->getStableTimestamp();
     ASSERT_EQUALS(Timestamp(2, 1), stableTimestamp);
 
     // Check that timestamp cleanup occurs.
-    repl->setMyLastAppliedOpTime(OpTime({2, 2}, term));
+    repl->setMyLastAppliedOpTime(OpTimeWithTermOne(2, 2));
     stableTimestamp = getStorageInterface()->getStableTimestamp();
     ASSERT_EQUALS(Timestamp(2, 2), stableTimestamp);
 
     auto opTimeCandidates = repl->getStableOpTimeCandidates_forTest();
-    std::set<OpTime> expectedOpTimeCandidates = {OpTime({2, 2}, term)};
+    std::set<OpTime> expectedOpTimeCandidates = {OpTimeWithTermOne(2, 2)};
     ASSERT_OPTIME_SET_EQ(expectedOpTimeCandidates, opTimeCandidates);
 }
 
@@ -4078,9 +4076,12 @@ TEST_F(StableOpTimeTest, ClearOpTimeCandidatesPastCommonPointAfterRollback) {
 
     OpTime rollbackCommonPoint = OpTime({1, 2}, term);
     OpTime commitPoint = OpTime({1, 2}, term);
-    repl->advanceCommitPoint(commitPoint);
     ASSERT_EQUALS(Timestamp::min(), getStorageInterface()->getStableTimestamp());
 
+    repl->setMyLastAppliedOpTime(OpTime({0, 1}, term));
+    // Advance commit point when it has the same term as the last applied.
+    repl->advanceCommitPoint(commitPoint);
+
     repl->setMyLastAppliedOpTime(OpTime({1, 1}, term));
     repl->setMyLastAppliedOpTime(OpTime({1, 2}, term));
     repl->setMyLastAppliedOpTime(OpTime({1, 3}, term));
@@ -4678,6 +4679,35 @@ TEST_F(ReplCoordTest,
     ASSERT_EQUALS(-1, getTopoCoord().getCurrentPrimaryIndex());
 }
 
+TEST_F(ReplCoordTest, AdvanceCommitPointFromSyncSourceCanSetCommitPointToLastAppliedIgnoringTerm) {
+    assertStartSuccess(BSON("_id"
+                            << "mySet"
+                            << "version"
+                            << 2
+                            << "members"
+                            << BSON_ARRAY(BSON("host"
+                                               << "node1:12345"
+                                               << "_id"
+                                               << 0)
+                                          << BSON("host"
+                                                  << "node2:12345"
+                                                  << "_id"
+                                                  << 1))
+                            << "protocolVersion"
+                            << 1),
+                       HostAndPort("node1", 12345));
+    ASSERT_EQUALS(OpTime(), getReplCoord()->getLastCommittedOpTime());
+
+    auto lastApplied = OpTime({10, 1}, 1);
+    auto commitPoint = OpTime({15, 1}, 2);
+    getReplCoord()->setMyLastAppliedOpTime(lastApplied);
+
+    getReplCoord()->advanceCommitPoint(commitPoint);
+
+    // The commit point can be set to lastApplied, even though lastApplied is in a lower term.
+    ASSERT_EQUALS(lastApplied, getReplCoord()->getLastCommittedOpTime());
+}
+
 TEST_F(ReplCoordTest, PrepareOplogQueryMetadata) {
     assertStartSuccess(BSON("_id"
                             << "mySet"
@@ -4701,11 +4731,11 @@ TEST_F(ReplCoordTest, PrepareOplogQueryMetadata) {
                        HostAndPort("node1", 12345));
     ASSERT_OK(getReplCoord()->setFollowerMode(MemberState::RS_SECONDARY));
 
-    OpTime optime1{Timestamp(10, 0), 3};
+    OpTime optime1{Timestamp(10, 0), 5};
     OpTime optime2{Timestamp(11, 2), 5};
 
-    getReplCoord()->advanceCommitPoint(optime1);
     getReplCoord()->setMyLastAppliedOpTime(optime2);
+    getReplCoord()->advanceCommitPoint(optime1);
 
     auto opCtx = makeOperationContext();
 
diff --git a/src/mongo/db/repl/topology_coordinator.cpp b/src/mongo/db/repl/topology_coordinator.cpp
index b6fb47ac5fa..7a448409550 100644
--- a/src/mongo/db/repl/topology_coordinator.cpp
+++ b/src/mongo/db/repl/topology_coordinator.cpp
@@ -3088,13 +3088,11 @@ bool TopologyCoordinator::updateLastCommittedOpTime() {
     return advanceLastCommittedOpTime(committedOpTime);
 }
 
-bool TopologyCoordinator::advanceLastCommittedOpTime(const OpTime& committedOpTime) {
-    if (committedOpTime == _lastCommittedOpTime) {
-        return false;  // Hasn't changed, so ignore it.
-    } else if (committedOpTime < _lastCommittedOpTime) {
-        LOG(1) << "Ignoring older committed snapshot optime: " << committedOpTime
-               << ", currentCommittedOpTime: " << _lastCommittedOpTime;
-        return false;  // This may have come from an out-of-order heartbeat. Ignore it.
+bool TopologyCoordinator::advanceLastCommittedOpTime(OpTime committedOpTime) {
+    if (_selfIndex == -1) {
+        // The config hasn't been installed or we are not in the config. This could happen
+        // on heartbeats before installing a config.
+        return false;
     }
 
     // This check is performed to ensure primaries do not commit an OpTime from a previous term.
@@ -3104,6 +3102,22 @@ bool TopologyCoordinator::advanceLastCommittedOpTime(const OpTime& committedOpTi
         return false;
     }
 
+    // Arbiters don't have data so they always advance their commit point via heartbeats.
+    if (!_selfConfig().isArbiter() &&
+        getMyLastAppliedOpTime().getTerm() != committedOpTime.getTerm()) {
+        committedOpTime = std::min(committedOpTime, getMyLastAppliedOpTime());
+    }
+
+    if (committedOpTime == _lastCommittedOpTime) {
+        return false;  // Hasn't changed, so ignore it.
+    }
+
+    if (committedOpTime < _lastCommittedOpTime) {
+        LOG(1) << "Ignoring older committed snapshot optime: " << committedOpTime
+               << ", currentCommittedOpTime: " << _lastCommittedOpTime;
+        return false;
+    }
+
     LOG(2) << "Updating _lastCommittedOpTime to " << committedOpTime;
     _lastCommittedOpTime = committedOpTime;
     return true;
diff --git a/src/mongo/db/repl/topology_coordinator.h b/src/mongo/db/repl/topology_coordinator.h
index feb1c351c43..711f10815f8 100644
--- a/src/mongo/db/repl/topology_coordinator.h
+++ b/src/mongo/db/repl/topology_coordinator.h
@@ -250,10 +250,10 @@ public:
     bool updateLastCommittedOpTime();
 
     /**
-     * Updates _lastCommittedOpTime to be "committedOpTime" if it is more recent than the
-     * current last committed OpTime.  Returns true if _lastCommittedOpTime is changed.
+     * Updates _lastCommittedOpTime to min(committedOpTime, lastApplied) if it is more recent than
+     * the current last committed OpTime.  Returns true if _lastCommittedOpTime is changed.
      */
-    bool advanceLastCommittedOpTime(const OpTime& committedOpTime);
+    bool advanceLastCommittedOpTime(OpTime committedOpTime);
 
     /**
      * Returns the OpTime of the latest majority-committed op known to this server.
diff --git a/src/mongo/db/repl/topology_coordinator_v1_test.cpp b/src/mongo/db/repl/topology_coordinator_v1_test.cpp
index a45f54805da..41dfcec5d57 100644
--- a/src/mongo/db/repl/topology_coordinator_v1_test.cpp
+++ b/src/mongo/db/repl/topology_coordinator_v1_test.cpp
@@ -1583,11 +1583,11 @@ TEST_F(TopoCoordTest, ReplSetGetStatus) {
     Seconds uptimeSecs(10);
     Date_t curTime = heartbeatTime + uptimeSecs;
     Timestamp electionTime(1, 2);
-    OpTime oplogProgress(Timestamp(3, 4), 2);
-    OpTime oplogDurable(Timestamp(3, 4), 1);
-    OpTime lastCommittedOpTime(Timestamp(2, 3), 6);
-    OpTime readConcernMajorityOpTime(Timestamp(4, 5), 7);
-    Timestamp lastStableCheckpointTimestamp(9, 9);
+    OpTime oplogProgress(Timestamp(3, 1), 20);
+    OpTime oplogDurable(Timestamp(1, 1), 19);
+    OpTime lastCommittedOpTime(Timestamp(5, 1), 20);
+    OpTime readConcernMajorityOpTime(Timestamp(4, 1), 20);
+    Timestamp lastStableCheckpointTimestamp(2, 2);
     BSONObj initialSyncStatus = BSON("failedInitialSyncAttempts" << 1);
     std::string setName = "mySet";
 
diff --git a/src/mongo/shell/replsettest.js b/src/mongo/shell/replsettest.js
index 796ffa642f1..ffa2ba9875c 100644
--- a/src/mongo/shell/replsettest.js
+++ b/src/mongo/shell/replsettest.js
@@ -1193,18 +1193,28 @@ var ReplSetTest = function(opts) {
      * of the oplog on *all* secondaries.
      * Returns last oplog entry.
      */
-    this.awaitLastOpCommitted = function(timeout) {
+    this.awaitLastOpCommitted = function(timeout, members) {
         var rst = this;
         var master = rst.getPrimary();
         var masterOpTime = _getLastOpTime(master);
 
-        print("Waiting for op with OpTime " + tojson(masterOpTime) +
-              " to be committed on all secondaries");
+        let membersToCheck;
+        if (members !== undefined) {
+            print("Waiting for op with OpTime " + tojson(masterOpTime) + " to be committed on " +
+                  members.map(s => s.host));
+
+            membersToCheck = members;
+        } else {
+            print("Waiting for op with OpTime " + tojson(masterOpTime) +
+                  " to be committed on all secondaries");
+
+            membersToCheck = rst.nodes;
+        }
 
         assert.soonNoExcept(
             function() {
-                for (var i = 0; i < rst.nodes.length; i++) {
-                    var node = rst.nodes[i];
+                for (var i = 0; i < membersToCheck.length; i++) {
+                    var node = membersToCheck[i];
 
                     // Continue if we're connected to an arbiter
                     var res = assert.commandWorked(node.adminCommand({replSetGetStatus: 1}));