Revert "SERVER-85697 Make initial sync and recovery use OplogWriter for writing oplog entries (#20676)"r8.0.0-rc0

This reverts commit 281edaac0072b57a288bab4d90b0d3151ff33a83.

GitOrigin-RevId: c732920bcf93f854d5dcd1999dfba18f9cc5ff42

13 files changed, 363 insertions, 352 deletions

diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index cfb4850e7cd..58f65d54d43 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -694,7 +694,6 @@ env.Library(
         '$BUILD_DIR/mongo/executor/task_executor_interface',
         '$BUILD_DIR/mongo/util/concurrency/thread_pool',
         'oplog_entry',
-        'repl_server_parameters',
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/admission/execution_admission_context',
@@ -705,6 +704,7 @@ env.Library(
         '$BUILD_DIR/mongo/util/fail_point',
         '$BUILD_DIR/mongo/util/processinfo',
         'repl_coordinator_interface',
+        'repl_server_parameters',
     ],
 )
 
@@ -727,7 +727,6 @@ env.Library(
         'oplog',
         'oplog_application_interface',
         'oplog_entry',
-        'oplog_write',
         'repl_coordinator_interface',
         'repl_settings',
         'replication_metrics',
diff --git a/src/mongo/db/repl/data_replicator_external_state_mock.cpp b/src/mongo/db/repl/data_replicator_external_state_mock.cpp
index b873e549058..ce6c818798e 100644
--- a/src/mongo/db/repl/data_replicator_external_state_mock.cpp
+++ b/src/mongo/db/repl/data_replicator_external_state_mock.cpp
@@ -60,6 +60,12 @@ public:
           _observer(observer),
           _externalState(externalState) {}
 
+    void scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+                                                  StorageInterface* storageInterface,
+                                                  ThreadPool* writerPool,
+                                                  const std::vector<OplogEntry>& ops,
+                                                  bool skipWritesToOplog) override;
+
 private:
     void _run(OplogBuffer* oplogBuffer) final {}
     StatusWith<OpTime> _applyOplogBatch(OperationContext* opCtx,
@@ -71,6 +77,12 @@ private:
     DataReplicatorExternalStateMock* const _externalState;
 };
 
+void OplogApplierMock::scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+                                                                StorageInterface* storageInterface,
+                                                                ThreadPool* writerPool,
+                                                                const std::vector<OplogEntry>& ops,
+                                                                bool skipWritesToOplog) {}
+
 }  // namespace
 
 DataReplicatorExternalStateMock::DataReplicatorExternalStateMock()
diff --git a/src/mongo/db/repl/oplog_application_bm.cpp b/src/mongo/db/repl/oplog_application_bm.cpp
index e15bd20b818..5777069175d 100644
--- a/src/mongo/db/repl/oplog_application_bm.cpp
+++ b/src/mongo/db/repl/oplog_application_bm.cpp
@@ -222,7 +222,8 @@ public:
         repl::OplogApplier::Options oplogApplierOptions(
             repl::OplogApplication::Mode::kSecondary,
             false /* allowNamespaceNotFoundErrorsOnCrudOps */,
-            true /* skipWritesToOplog */);
+            true /* skipWritesToOplog */,
+            true /* skipWritesToChangeCollection */);
         _oplogApplier = std::make_unique<repl::OplogApplierImpl>(nullptr,
                                                                  _oplogBuffer.get(),
                                                                  &repl::noopOplogApplierObserver,
diff --git a/src/mongo/db/repl/oplog_applier.h b/src/mongo/db/repl/oplog_applier.h
index 0ddd049cb55..dab7776ae2b 100644
--- a/src/mongo/db/repl/oplog_applier.h
+++ b/src/mongo/db/repl/oplog_applier.h
@@ -77,25 +77,27 @@ public:
               allowNamespaceNotFoundErrorsOnCrudOps(inputMode ==
                                                         OplogApplication::Mode::kInitialSync ||
                                                     OplogApplication::inRecovering(inputMode)),
-              skipWritesToOplog(
-                  (feature_flags::gReduceMajorityWriteLatency.isEnabled(
-                       serverGlobalParams.featureCompatibility.acquireFCVSnapshot()) &&
-                   inputMode == OplogApplication::Mode::kSecondary) ||
-                  OplogApplication::inRecovering(inputMode)) {}
+              skipWritesToOplog(OplogApplication::inRecovering(inputMode)),
+              skipWritesToChangeCollection(false) {}
 
-        Options(OplogApplication::Mode inputMode, bool skipWritesToOplog)
+        Options(OplogApplication::Mode inputMode,
+                bool skipWritesToOplog,
+                bool skipWritesToChangeCollection)
             : mode(inputMode),
               allowNamespaceNotFoundErrorsOnCrudOps(inputMode ==
                                                         OplogApplication::Mode::kInitialSync ||
                                                     OplogApplication::inRecovering(inputMode)),
-              skipWritesToOplog(skipWritesToOplog) {}
+              skipWritesToOplog(skipWritesToOplog),
+              skipWritesToChangeCollection(skipWritesToChangeCollection) {}
 
         Options(OplogApplication::Mode mode,
                 bool allowNamespaceNotFoundErrorsOnCrudOps,
-                bool skipWritesToOplog)
+                bool skipWritesToOplog,
+                bool skipWritesToChangeCollection)
             : mode(mode),
               allowNamespaceNotFoundErrorsOnCrudOps(allowNamespaceNotFoundErrorsOnCrudOps),
-              skipWritesToOplog(skipWritesToOplog) {}
+              skipWritesToOplog(skipWritesToOplog),
+              skipWritesToChangeCollection(skipWritesToChangeCollection) {}
 
         // Used to determine which operations should be applied. Only initial sync will set this to
         // be something other than the null optime.
@@ -104,6 +106,7 @@ public:
         const OplogApplication::Mode mode;
         const bool allowNamespaceNotFoundErrorsOnCrudOps;
         const bool skipWritesToOplog;
+        const bool skipWritesToChangeCollection;
     };
 
     // Used to report oplog application progress.
@@ -183,6 +186,12 @@ public:
      */
     StatusWith<OpTime> applyOplogBatch(OperationContext* opCtx, std::vector<OplogEntry> ops);
 
+    virtual void scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+                                                          StorageInterface* storageInterface,
+                                                          ThreadPool* writerPool,
+                                                          const std::vector<OplogEntry>& ops,
+                                                          bool skipWritesToOplog) = 0;
+
     /**
      * Calls the OplogBatcher's getNextApplierBatch.
      */
diff --git a/src/mongo/db/repl/oplog_applier_impl.cpp b/src/mongo/db/repl/oplog_applier_impl.cpp
index 56d562b29bb..d02aba3e0c2 100644
--- a/src/mongo/db/repl/oplog_applier_impl.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl.cpp
@@ -67,7 +67,6 @@
 #include "mongo/db/repl/oplog_applier_utils.h"
 #include "mongo/db/repl/oplog_batcher.h"
 #include "mongo/db/repl/oplog_entry_gen.h"
-#include "mongo/db/repl/oplog_writer_impl.h"
 #include "mongo/db/repl/replication_metrics.h"
 #include "mongo/db/repl/split_prepare_session_manager.h"
 #include "mongo/db/repl/transaction_oplog_application.h"
@@ -342,6 +341,58 @@ void _addOplogChainOpsToWriterVectors(OperationContext* opCtx,
         opCtx, &extractedOps, writerVectors, collPropertiesCache, shouldSerialize);
 }
 
+Status _insertDocumentsToOplogAndChangeCollections(
+    OperationContext* opCtx,
+    std::vector<InsertStatement>::const_iterator begin,
+    std::vector<InsertStatement>::const_iterator end,
+    bool skipWritesToOplog) {
+    WriteUnitOfWork wunit(opCtx);
+    boost::optional<AutoGetOplogFastPath> autoOplog;
+    boost::optional<ChangeStreamChangeCollectionManager::ChangeCollectionsWriter>
+        changeCollectionWriter;
+
+    // Acquire locks. We must acquire the locks for all collections we intend to write to before
+    // performing any writes. This avoids potential deadlocks created by waiting for locks while
+    // having generated oplog holes.
+    if (!skipWritesToOplog) {
+        autoOplog.emplace(opCtx, OplogAccessMode::kWrite);
+    }
+    const bool changeCollectionsMode =
+        change_stream_serverless_helpers::isChangeCollectionsModeActive();
+    if (changeCollectionsMode) {
+        changeCollectionWriter = boost::make_optional(
+            ChangeStreamChangeCollectionManager::get(opCtx).createChangeCollectionsWriter(
+                opCtx, begin, end, nullptr /* opDebug */));
+        changeCollectionWriter->acquireLocks();
+    }
+
+    // Write entries to the oplog.
+    if (!skipWritesToOplog) {
+        auto& oplogColl = autoOplog->getCollection();
+        if (!oplogColl) {
+            return {ErrorCodes::NamespaceNotFound, "Oplog collection does not exist"};
+        }
+        auto status = collection_internal::insertDocuments(
+            opCtx, oplogColl, begin, end, nullptr /* OpDebug */, false /* fromMigrate */);
+        if (!status.isOK()) {
+            return status;
+        }
+    }
+
+    // Write the corresponding oplog entries to tenants respective change
+    // collections in the serverless.
+    if (changeCollectionsMode) {
+        auto status = changeCollectionWriter->write();
+        if (!status.isOK()) {
+            return status;
+        }
+    }
+
+    wunit.commit();
+
+    return Status::OK();
+}
+
 void _setOplogApplicationWorkerOpCtxStates(OperationContext* opCtx) {
     // Do not enforce constraints.
     opCtx->setEnforceConstraints(false);
@@ -476,23 +527,8 @@ OplogApplierImpl::OplogApplierImpl(executor::TaskExecutor* executor,
       _replCoord(replCoord),
       _writerPool(writerPool),
       _storageInterface(storageInterface),
-      _consistencyMarkers(consistencyMarkers) {
-
-    // Change collections are always written as part of oplog application, even though
-    // in steady state mode where a separate OplogWriter thread is responsible for
-    // writing oplog entries. This is because the OplogWriter thread can be ahead of
-    // oplog application, but DDL ops on change collections (e.g. create/drop) must be
-    // first applied before we can perform writes on those collections.
-    _oplogWriter = std::make_unique<OplogWriterImpl>(
-        nullptr /* executor */,
-        nullptr /* writeBuffer */,
-        nullptr /* applyBuffer */,
-        replCoord,
-        storageInterface,
-        consistencyMarkers,
-        &noopOplogWriterObserver,
-        OplogWriter::Options(options.skipWritesToOplog, false /* skipWritesToChangeColl */));
-}
+      _consistencyMarkers(consistencyMarkers),
+      _beginApplyingOpTime(options.beginApplyingOpTime) {}
 
 void OplogApplierImpl::_run(OplogBuffer* oplogBuffer) {
     // Start up a thread from the batcher to pull from the oplog buffer into the batcher's oplog
@@ -618,6 +654,86 @@ void OplogApplierImpl::_run(OplogBuffer* oplogBuffer) {
     }
 }
 
+
+// Schedules the writes to the oplog and the change collection for 'ops' into threadPool. The caller
+// must guarantee that 'ops' stays valid until all scheduled work in the thread pool completes.
+void OplogApplierImpl::scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+                                                                StorageInterface* storageInterface,
+                                                                ThreadPool* writerPool,
+                                                                const std::vector<OplogEntry>& ops,
+                                                                bool skipWritesToOplog) {
+    // Skip performing any writes during the startup recovery when running in the non-serverless
+    // environment.
+    if (skipWritesToOplog && !change_stream_serverless_helpers::isChangeCollectionsModeActive()) {
+        return;
+    }
+
+    auto makeOplogWriterForRange = [storageInterface, &ops, skipWritesToOplog](size_t begin,
+                                                                               size_t end) {
+        // The returned function will be run in a separate thread after this returns. Therefore
+        // all captures other than 'ops' must be by value since they will not be available. The
+        // caller guarantees that 'ops' will stay in scope until the spawned threads complete.
+        return [storageInterface, &ops, begin, end, skipWritesToOplog](auto status) {
+            invariant(status);
+            auto opCtx = cc().makeOperationContext();
+
+            // Oplog writes are crucial to the stability of the replica set. We mark the operations
+            // as having Immediate priority so that it skips waiting for ticket acquisition and flow
+            // control.
+            ScopedAdmissionPriority<ExecutionAdmissionContext> priority(
+                opCtx.get(), AdmissionContext::Priority::kExempt);
+
+            UnreplicatedWritesBlock uwb(opCtx.get());
+
+            std::vector<InsertStatement> docs;
+            docs.reserve(end - begin);
+            for (size_t i = begin; i < end; i++) {
+                docs.emplace_back(InsertStatement{ops[i].getEntry().getRaw(),
+                                                  ops[i].getOpTime().getTimestamp(),
+                                                  ops[i].getOpTime().getTerm()});
+            }
+
+            // The 'nsOrUUID' is used only to log the debug message when retrying inserts on the
+            // oplog and change collections. The 'writeConflictRetry' assumes operations are done on
+            // a single namespace. But the method '_insertDocumentsToOplogAndChangeCollections' can
+            // perform inserts on the oplog and multiple change collections, ie. several namespaces.
+            // As such 'writeConflictRetry' will not log the correct namespace when retrying.
+            NamespaceStringOrUUID nsOrUUID = !skipWritesToOplog
+                ? NamespaceString::kRsOplogNamespace
+                : NamespaceString::makeChangeCollectionNSS(boost::none /* tenantId */);
+
+            fassert(6663400,
+                    storage_helpers::insertBatchAndHandleRetry(
+                        opCtx.get(), nsOrUUID, docs, [&](auto* opCtx, auto begin, auto end) {
+                            return _insertDocumentsToOplogAndChangeCollections(
+                                opCtx, begin, end, skipWritesToOplog);
+                        }));
+        };
+    };
+
+    // We want to be able to take advantage of bulk inserts so we don't use multiple threads if it
+    // would result too little work per thread. This also ensures that we can amortize the
+    // setup/teardown overhead across many writes.
+    const size_t kMinOplogEntriesPerThread = 16;
+    const bool enoughToMultiThread =
+        ops.size() >= kMinOplogEntriesPerThread * writerPool->getStats().options.maxThreads;
+
+    // Storage engines support parallel writes to the oplog because they are required to ensure that
+    // oplog entries are ordered correctly, even if inserted out-of-order.
+    if (!enoughToMultiThread) {
+        writerPool->schedule(makeOplogWriterForRange(0, ops.size()));
+        return;
+    }
+
+    const size_t numOplogThreads = writerPool->getStats().options.maxThreads;
+    const size_t numOpsPerThread = ops.size() / numOplogThreads;
+    for (size_t thread = 0; thread < numOplogThreads; thread++) {
+        size_t begin = thread * numOpsPerThread;
+        size_t end = (thread == numOplogThreads - 1) ? ops.size() : begin + numOpsPerThread;
+        writerPool->schedule(makeOplogWriterForRange(begin, end));
+    }
+}
+
 StatusWith<OpTime> OplogApplierImpl::_applyOplogBatch(OperationContext* opCtx,
                                                       std::vector<OplogEntry> ops) {
     invariant(!ops.empty());
@@ -642,19 +758,16 @@ StatusWith<OpTime> OplogApplierImpl::_applyOplogBatch(OperationContext* opCtx,
         // because the spawned threads refer to objects on the stack
         ON_BLOCK_EXIT([&] { _writerPool->waitForIdle(); });
 
-        // Write ops into the oplog collection and change collections if needed.
-        //
-        // - In steady state mode, a separate OplogWriter is responsible for writing the oplog
-        // collection, but the change collections must be written as part of oplog application
-        // because DDL ops (e.g. create/drop) on change collections are explicitly replicated,
-        // and so must be first applied before writes are performed to those collections.
-        //
-        // - In initial sync mode, the applier is responsible for writing the oplog collection
-        // as well as the change collections.
-        //
-        // - In recovering modes, there is no need to write to the oplog collection, and the
-        // applier is responsible for writing the change collections.
-        _oplogWriter->writeOplogBatch(opCtx, ops, _writerPool);
+        // Write batch of ops into oplog.
+        if (!getOptions().skipWritesToOplog) {
+            _consistencyMarkers->setOplogTruncateAfterPoint(
+                opCtx, _replCoord->getMyLastAppliedOpTime().getTimestamp());
+        }
+
+        if (!getOptions().skipWritesToOplog || !getOptions().skipWritesToChangeCollection) {
+            scheduleWritesToOplogAndChangeCollection(
+                opCtx, _storageInterface, _writerPool, ops, getOptions().skipWritesToOplog);
+        }
 
         // Holds 'pseudo operations' generated by secondaries to aid in replication.
         // Keep in scope until all operations in 'ops' and 'derivedOps' have been applied.
@@ -669,6 +782,9 @@ StatusWith<OpTime> OplogApplierImpl::_applyOplogBatch(OperationContext* opCtx,
             _writerPool->getStats().options.maxThreads);
         _fillWriterVectors(opCtx, &ops, &writerVectors, &derivedOps);
 
+        // Wait for writes to finish before applying ops.
+        _writerPool->waitForIdle();
+
         // Use this fail point to hang after we have written the oplog entries but before we have
         // applied them.
         if (MONGO_unlikely(pauseBatchApplicationAfterWritingOplogEntries.shouldFail())) {
@@ -682,6 +798,11 @@ StatusWith<OpTime> OplogApplierImpl::_applyOplogBatch(OperationContext* opCtx,
         const bool isDataConsistent =
             _consistencyMarkers->getMinValid(opCtx) < ops.front().getOpTime();
 
+        // Reset consistency markers in case the node fails while applying ops.
+        if (!getOptions().skipWritesToOplog) {
+            _consistencyMarkers->setOplogTruncateAfterPoint(opCtx, Timestamp());
+        }
+
         {
             std::vector<Status> statusVector(_writerPool->getStats().options.maxThreads,
                                              Status::OK());
diff --git a/src/mongo/db/repl/oplog_applier_impl.h b/src/mongo/db/repl/oplog_applier_impl.h
index f6ab46a6e20..dae3b9b7361 100644
--- a/src/mongo/db/repl/oplog_applier_impl.h
+++ b/src/mongo/db/repl/oplog_applier_impl.h
@@ -44,7 +44,6 @@
 #include "mongo/db/repl/oplog_buffer.h"
 #include "mongo/db/repl/oplog_entry.h"
 #include "mongo/db/repl/oplog_entry_or_grouped_inserts.h"
-#include "mongo/db/repl/oplog_writer.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/replication_consistency_markers.h"
 #include "mongo/db/repl/replication_coordinator.h"
@@ -92,6 +91,12 @@ public:
                                    std::vector<std::vector<ApplierOperation>>* writerVectors,
                                    std::vector<std::vector<OplogEntry>>* derivedOps) noexcept;
 
+    void scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+                                                  StorageInterface* storageInterface,
+                                                  ThreadPool* writerPool,
+                                                  const std::vector<OplogEntry>& ops,
+                                                  bool skipWritesToOplog) override;
+
 private:
     /**
      * Runs oplog application in a loop until shutdown() is called.
@@ -134,11 +139,13 @@ private:
     // Not owned by us.
     ThreadPool* const _writerPool;
 
-    StorageInterface* const _storageInterface;
+    StorageInterface* _storageInterface;
 
     ReplicationConsistencyMarkers* const _consistencyMarkers;
 
-    std::unique_ptr<OplogWriter> _oplogWriter;
+    // Used to determine which operations should be applied during initial sync. If this is null,
+    // we will apply all operations that were fetched.
+    OpTime _beginApplyingOpTime = OpTime();
 
 protected:
     // Marked as protected for use in unit tests.
diff --git a/src/mongo/db/repl/oplog_applier_test.cpp b/src/mongo/db/repl/oplog_applier_test.cpp
index 56d1796fe62..eddf69ff8f7 100644
--- a/src/mongo/db/repl/oplog_applier_test.cpp
+++ b/src/mongo/db/repl/oplog_applier_test.cpp
@@ -73,6 +73,11 @@ public:
 
     void _run(OplogBuffer* oplogBuffer) final;
     StatusWith<OpTime> _applyOplogBatch(OperationContext* opCtx, std::vector<OplogEntry> ops) final;
+    void scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+                                                  StorageInterface* storageInterface,
+                                                  ThreadPool* writerPool,
+                                                  const std::vector<OplogEntry>& ops,
+                                                  bool skipWritesToOplog) override;
 };
 
 OplogApplierMock::OplogApplierMock(OplogBuffer* oplogBuffer)
@@ -88,6 +93,12 @@ StatusWith<OpTime> OplogApplierMock::_applyOplogBatch(OperationContext* opCtx,
     return OpTime();
 }
 
+void OplogApplierMock::scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+                                                                StorageInterface* storageInterface,
+                                                                ThreadPool* writerPool,
+                                                                const std::vector<OplogEntry>& ops,
+                                                                bool skipWritesToOplog) {}
+
 class OplogApplierTest : public ServiceContextTest {
 public:
     void setUp() override;
diff --git a/src/mongo/db/repl/oplog_write_bm.cpp b/src/mongo/db/repl/oplog_write_bm.cpp
index eecb7579835..479911fcc9f 100644
--- a/src/mongo/db/repl/oplog_write_bm.cpp
+++ b/src/mongo/db/repl/oplog_write_bm.cpp
@@ -78,7 +78,6 @@
 #include "mongo/db/repl/oplog_buffer.h"
 #include "mongo/db/repl/oplog_buffer_blocking_queue.h"
 #include "mongo/db/repl/oplog_entry.h"
-#include "mongo/db/repl/oplog_writer_impl.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/repl_settings.h"
 #include "mongo/db/repl/replication_consistency_markers_mock.h"
@@ -208,21 +207,26 @@ public:
             std::make_unique<MongoDSessionCatalog>(
                 std::make_unique<MongoDSessionCatalogTransactionInterfaceImpl>()));
 
+        _oplogBuffer = std::make_unique<repl::OplogBufferBlockingQueue>(kOplogBufferSize);
+
         repl::replWriterThreadCount = numThreads;  // Repl worker thread count
         repl::replWriterMinThreadCount = numThreads;
 
-        _oplogWriterPool = repl::makeReplWriterPool();
-
-        _oplogWriter = std::make_unique<repl::OplogWriterImpl>(
-            nullptr,
-            nullptr,
-            nullptr,
-            _replCoord,
-            _storageInterface,
-            &_consistencyMarkers,
-            &repl::noopOplogWriterObserver,
-            repl::OplogWriter::Options(false /* skipWritesToOplogColl */,
-                                       true /* skipWritesToChangeColl */));
+        _oplogApplierThreadPool = repl::makeReplWriterPool();
+
+        repl::OplogApplier::Options oplogApplierOptions(
+            repl::OplogApplication::Mode::kSecondary,
+            false /* allowNamespaceNotFoundErrorsOnCrudOps */,
+            false /* skipWritesToOplog */,
+            false /* skipWritesToChangeCollection */);
+        _oplogApplier = std::make_unique<repl::OplogApplierImpl>(nullptr,
+                                                                 _oplogBuffer.get(),
+                                                                 &repl::noopOplogApplierObserver,
+                                                                 _replCoord,
+                                                                 &_consistencyMarkers,
+                                                                 _storageInterface,
+                                                                 oplogApplierOptions,
+                                                                 _oplogApplierThreadPool.get());
 
         _svcCtx->notifyStorageStartupRecoveryComplete();
     }
@@ -289,12 +293,12 @@ public:
         return _replCoord;
     }
 
-    repl::OplogWriter* getOplogWriter() {
-        return _oplogWriter.get();
+    repl::OplogApplier* getOplogApplier() {
+        return _oplogApplier.get();
     }
 
     ThreadPool* getThreadPool() {
-        return _oplogWriterPool.get();
+        return _oplogApplierThreadPool.get();
     }
 
     repl::StorageInterface* getStorageInterface() {
@@ -305,10 +309,13 @@ private:
     ServiceContext* _svcCtx;
     Client* _client;
     repl::ReplicationCoordinatorMock* _replCoord;
-    std::unique_ptr<repl::OplogWriter> _oplogWriter;
+    std::unique_ptr<repl::OplogApplier> _oplogApplier;
     repl::StorageInterface* _storageInterface;
+
+    // This class also owns objects necessary for `_oplogApplier`.
+    std::unique_ptr<repl::OplogBufferBlockingQueue> _oplogBuffer;
     repl::ReplicationConsistencyMarkersMock _consistencyMarkers;
-    std::unique_ptr<ThreadPool> _oplogWriterPool;
+    std::unique_ptr<ThreadPool> _oplogApplierThreadPool;
     boost::optional<unittest::TempDir> _tempDir;
 };
 
@@ -322,7 +329,7 @@ class Fixture {
 public:
     Fixture(TestServiceContext* testSvcCtx) : _testSvcCtx(testSvcCtx), _foobarUUID(UUID::gen()) {}
 
-    void generateEntries(int totalOps, int entrySize) {
+    void createBatch(int totalOps, int entrySize) {
         const long long term1 = 1;
         for (int idx = 0; idx < totalOps; ++idx) {
             auto x1 =
@@ -375,40 +382,27 @@ public:
         }
     }
 
-    std::vector<std::vector<BSONObj>> getOplogBatches(size_t numEntriesPerBatch,
-                                                      size_t numBytesPerBatch) {
-        invariant(!_oplogEntries.empty());
-
-        std::vector<std::vector<BSONObj>> batches;
-        std::vector<BSONObj>::iterator first = _oplogEntries.begin();
-
-        std::size_t batchEntries = 1;
-        std::size_t batchBytes = first->objsize();
-
-        for (auto last = first + 1; last != _oplogEntries.end(); ++last) {
-            // Create a new batch if batch limits exceeded.
-            if ((batchEntries + 1 > numEntriesPerBatch) ||
-                (batchBytes + last->objsize() > numBytesPerBatch)) {
-                batches.emplace_back(first, last);
-                batchEntries = 1;
-                batchBytes = last->objsize();
-                first = last;
-                continue;
-            }
-            // Otherwise update batch stats and move on.
-            ++batchEntries;
-            batchBytes += last->objsize();
-        }
-        batches.emplace_back(first, _oplogEntries.end());
-
-        return batches;
+    void enqueueOplog(OperationContext* opCtx) {
+        _testSvcCtx->getOplogApplier()->enqueue(opCtx, _oplogEntries.begin(), _oplogEntries.end());
     }
 
-    void writeOplog(OperationContext* opCtx, const std::vector<std::vector<BSONObj>>& batches) {
-        for (const auto& batch : batches) {
+    void writeOplog(OperationContext* opCtx, size_t numEntriesPerBatch, size_t numBytesPerBatch) {
+        while (!_testSvcCtx->getOplogApplier()->getBuffer()->isEmpty()) {
+            auto oplogBatch = invariantStatusOK(_testSvcCtx->getOplogApplier()->getNextApplierBatch(
+                                                    opCtx, {numBytesPerBatch, numEntriesPerBatch}))
+                                  .releaseBatch();
+
             AutoGetDb autoDb(opCtx, _foobarNs.dbName(), MODE_X);
-            invariant(_testSvcCtx->getOplogWriter()->writeOplogBatch(
-                opCtx, batch, _testSvcCtx->getThreadPool()));
+
+            _testSvcCtx->getOplogApplier()->scheduleWritesToOplogAndChangeCollection(
+                opCtx,
+                _testSvcCtx->getStorageInterface(),
+                _testSvcCtx->getThreadPool(),
+                std::move(oplogBatch),
+                false);
+
+            // Wait for writes to finish
+            _testSvcCtx->getThreadPool()->waitForIdle();
         }
     }
 
@@ -427,9 +421,9 @@ void runBMTest(TestServiceContext& testSvcCtx, Fixture& fixture, benchmark::Stat
         auto opCtxRaii = testSvcCtx.getSvcCtx()->makeOperationContext(testSvcCtx.getClient());
         auto opCtx = opCtxRaii.get();
         repl::createOplog(opCtx);
-        auto batches = fixture.getOplogBatches(state.range(3), state.range(4));
+        fixture.enqueueOplog(opCtx);
         auto start = mongo::stdx::chrono::high_resolution_clock::now();
-        fixture.writeOplog(opCtx, batches);
+        fixture.writeOplog(opCtx, state.range(3), state.range(4));
         auto end = mongo::stdx::chrono::high_resolution_clock::now();
         auto elapsed_seconds =
             mongo::stdx::chrono::duration_cast<mongo::stdx::chrono::duration<double>>(end - start);
@@ -440,7 +434,7 @@ void runBMTest(TestServiceContext& testSvcCtx, Fixture& fixture, benchmark::Stat
 void BM_TestWriteOps(benchmark::State& state) {
     TestServiceContext testSvcCtx(state.range(0));
     Fixture fixture(&testSvcCtx);
-    fixture.generateEntries(state.range(1), state.range(2));
+    fixture.createBatch(state.range(1), state.range(2));
     runBMTest(testSvcCtx, fixture, state);
 }
 
diff --git a/src/mongo/db/repl/oplog_writer.h b/src/mongo/db/repl/oplog_writer.h
index 4a0681ae98a..401289261fb 100644
--- a/src/mongo/db/repl/oplog_writer.h
+++ b/src/mongo/db/repl/oplog_writer.h
@@ -33,7 +33,6 @@
 #include "mongo/db/repl/oplog_buffer.h"
 #include "mongo/db/repl/oplog_writer_batcher.h"
 #include "mongo/executor/task_executor.h"
-#include "mongo/util/concurrency/thread_pool.h"
 
 namespace mongo {
 namespace repl {
@@ -51,13 +50,11 @@ public:
      */
     class Options {
     public:
-        Options() = delete;
-        explicit Options(bool skipWritesToOplogColl, bool skipWritesToChangeColl)
-            : skipWritesToOplogColl(skipWritesToOplogColl),
-              skipWritesToChangeColl(skipWritesToChangeColl) {}
+        Options() : skipWritesToOplogColl(false) {}
+        explicit Options(bool skipWritesToOplogColl)
+            : skipWritesToOplogColl(skipWritesToOplogColl) {}
 
         const bool skipWritesToOplogColl;
-        const bool skipWritesToChangeColl;
     };
 
     // Used to report oplog write progress.
@@ -106,27 +103,17 @@ public:
                  boost::optional<std::size_t> bytes = boost::none);
 
     /**
-     * Writes a batch of oplog entries to the oplog and/or the change collections.
+     * Writes a batch of oplog entries to the oplog and/or the change collection.
      *
-     * Returns false if nothing is written, true otherwise.
+     * If the batch write is successful, returns the optime of the last op written,
+     * which should be the last op in the batch.
      *
      * Oplog visibility and updates to replication coordinator timestamps should be
      * handled by caller.
      */
-    virtual bool writeOplogBatch(OperationContext* opCtx,
-                                 const std::vector<BSONObj>& ops,
-                                 ThreadPool* writerPool = nullptr) = 0;
+    virtual StatusWith<OpTime> writeOplogBatch(OperationContext* opCtx,
+                                               const std::vector<BSONObj>& ops) = 0;
 
-    /**
-     * Same as above, except for the type of the oplog entries.
-     */
-    virtual bool writeOplogBatch(OperationContext* opCtx,
-                                 const std::vector<OplogEntry>& ops,
-                                 ThreadPool* writerPool = nullptr) = 0;
-
-    /**
-     * Returns the options used to configure the behavior of this OplogWriter.
-     */
     const Options& getOptions() const;
 
 private:
diff --git a/src/mongo/db/repl/oplog_writer_impl.cpp b/src/mongo/db/repl/oplog_writer_impl.cpp
index fb2c2b21bc3..e2595d78bff 100644
--- a/src/mongo/db/repl/oplog_writer_impl.cpp
+++ b/src/mongo/db/repl/oplog_writer_impl.cpp
@@ -46,7 +46,7 @@ namespace repl {
 
 namespace {
 
-constexpr size_t kMinOpsPerThread = 16;
+const auto changeCollNss = NamespaceString::makeChangeCollectionNSS(boost::none);
 
 auto checkFeatureFlagReduceMajorityWriteLatencyFn = [] {
     return feature_flags::gReduceMajorityWriteLatency.isEnabled(
@@ -106,35 +106,6 @@ Status insertDocsToOplogAndChangeCollections(OperationContext* opCtx,
     return Status::OK();
 }
 
-std::vector<InsertStatement> createInsertStatements(const std::vector<BSONObj>& ops,
-                                                    size_t begin,
-                                                    size_t end) {
-    std::vector<InsertStatement> docs;
-    docs.reserve(end - begin);
-
-    for (size_t i = begin; i < end; ++i) {
-        auto opTime = invariantStatusOK(OpTime::parseFromOplogEntry(ops[i]));
-        docs.emplace_back(ops[i], opTime.getTimestamp(), opTime.getTerm());
-    }
-
-    return docs;
-}
-
-std::vector<InsertStatement> createInsertStatements(const std::vector<repl::OplogEntry>& ops,
-                                                    size_t begin,
-                                                    size_t end) {
-    std::vector<InsertStatement> docs;
-    docs.reserve(end - begin);
-
-    for (size_t i = begin; i < end; ++i) {
-        docs.emplace_back(ops[i].getEntry().getRaw(),
-                          ops[i].getOpTime().getTimestamp(),
-                          ops[i].getOpTime().getTerm());
-    }
-
-    return docs;
-}
-
 }  // namespace
 
 
@@ -158,14 +129,12 @@ OplogWriterImpl::OplogWriterImpl(executor::TaskExecutor* executor,
                                  OplogBuffer* applyBuffer,
                                  ReplicationCoordinator* replCoord,
                                  StorageInterface* storageInterface,
-                                 ReplicationConsistencyMarkers* consistencyMarkers,
                                  Observer* observer,
                                  const OplogWriter::Options& options)
     : OplogWriter(executor, writeBuffer, options),
       _applyBuffer(applyBuffer),
       _replCoord(replCoord),
       _storageInterface(storageInterface),
-      _consistencyMarkers(consistencyMarkers),
       _observer(observer) {}
 
 void OplogWriterImpl::_run() {
@@ -219,16 +188,14 @@ void OplogWriterImpl::_run() {
         auto lastOpTimeAndWallTime =
             invariantStatusOK(OpTimeAndWallTime::parseOpTimeAndWallTimeFromOplogEntry(ops.back()));
 
-        {
-            LOGV2_DEBUG(8352100, 2, "Oplog write batch size", "size"_attr = ops.size());
-
-            // Increment the batch stats.
-            oplogWriterMetric.incrementBatchSize(ops.size());
-            TimerHolder timer(&oplogWriterMetric.getBatches());
-
-            // Write the operations in this batch.
-            invariant(writeOplogBatch(opCtx, ops));
+        // Write the operations in this batch. 'writeOplogBatch' returns the optime of
+        // the last op that was written, which should be the last optime in the batch.
+        auto swLastOpTime = writeOplogBatch(opCtx, ops);
+        if (swLastOpTime.getStatus().code() == ErrorCodes::InterruptedAtShutdown) {
+            return;
         }
+        fassertNoTrace(8543103, swLastOpTime);
+        invariant(swLastOpTime.getValue() == lastOpTimeAndWallTime.opTime);
 
         // Update various things that care about our last written optime.
         finalizeOplogBatch(opCtx, lastOpTimeAndWallTime, flushJournal);
@@ -238,16 +205,36 @@ void OplogWriterImpl::_run() {
     }
 }
 
-bool OplogWriterImpl::writeOplogBatch(OperationContext* opCtx,
-                                      const std::vector<BSONObj>& ops,
-                                      ThreadPool* writerPool) {
-    return _writeOplogBatch(opCtx, ops, writerPool);
-}
+StatusWith<OpTime> OplogWriterImpl::writeOplogBatch(OperationContext* opCtx,
+                                                    const std::vector<BSONObj>& ops) {
+    invariant(!ops.empty());
+    LOGV2_DEBUG(8352100, 2, "Oplog write batch size", "size"_attr = ops.size());
+
+    bool writeOplogColl = !getOptions().skipWritesToOplogColl;
+    bool writeChangeColl = change_stream_serverless_helpers::isChangeCollectionsModeActive();
+
+    // This should only happen for recovery modes.
+    if (!writeOplogColl && !writeChangeColl) {
+        return OpTime();
+    }
+
+    // Increment the batch stats.
+    oplogWriterMetric.incrementBatchSize(ops.size());
+    TimerHolder timer(&oplogWriterMetric.getBatches());
 
-bool OplogWriterImpl::writeOplogBatch(OperationContext* opCtx,
-                                      const std::vector<OplogEntry>& ops,
-                                      ThreadPool* writerPool) {
-    return _writeOplogBatch(opCtx, ops, writerPool);
+    // Create insert statements from the oplog entries.
+    std::vector<InsertStatement> docs;
+    docs.reserve(ops.size());
+
+    for (const auto& op : ops) {
+        auto opTime = invariantStatusOK(OpTime::parseFromOplogEntry(op));
+        docs.emplace_back(InsertStatement{op, opTime.getTimestamp(), opTime.getTerm()});
+    }
+
+    // Perform writes to oplog collection and/or change collection.
+    _writeOplogBatchImpl(opCtx, docs, writeOplogColl, writeChangeColl);
+
+    return docs.back().oplogSlot;
 }
 
 void OplogWriterImpl::finalizeOplogBatch(OperationContext* opCtx,
@@ -269,84 +256,16 @@ void OplogWriterImpl::finalizeOplogBatch(OperationContext* opCtx,
     }
 }
 
-template <typename T>
-bool OplogWriterImpl::_writeOplogBatch(OperationContext* opCtx,
-                                       const std::vector<T>& ops,
-                                       ThreadPool* writerPool) {
-    invariant(!ops.empty());
-
-    bool writeOplogColl = !getOptions().skipWritesToOplogColl;
-    bool writeChangeColl = !getOptions().skipWritesToChangeColl &&
-        change_stream_serverless_helpers::isChangeCollectionsModeActive();
-
-    // Don't do anything if not writing to the oplog collection nor the change collections.
-    if (!writeOplogColl && !writeChangeColl) {
-        return false;
-    }
-
-    // Perform writes to oplog collection and/or change collections with the current thread.
-    if (!writerPool) {
-        _writeOplogBatchForRange(opCtx, ops, 0, ops.size(), writeOplogColl, writeChangeColl);
-        return true;
-    }
-
-    // Perform writes to oplog collection and/or change collections using the thread pool.
-
-    // When performing writes with multiple threads, we must set oplogTruncateAfterPoint
-    // in case the server crashes before all the threads finish. In such cases the oplog
-    // will be truncated after this opTime during startup recovery in order to make sure
-    // there are no holes in the oplog.
-    if (writeOplogColl) {
-        _consistencyMarkers->setOplogTruncateAfterPoint(
-            opCtx, _replCoord->getMyLastWrittenOpTime().getTimestamp());
-    }
-
-    auto makeOplogWriteForRange = [this, &ops, writeOplogColl, writeChangeColl](size_t begin,
-                                                                                size_t end) {
-        return [this, &ops, begin, end, writeOplogColl, writeChangeColl](auto status) {
-            invariant(status);
-            auto opCtx = cc().makeOperationContext();
-            _writeOplogBatchForRange(opCtx.get(), ops, begin, end, writeOplogColl, writeChangeColl);
-        };
-    };
-
-    const auto poolMaxThreads = writerPool->getStats().options.maxThreads;
-    const auto enoughToMultiThread = ops.size() >= kMinOpsPerThread * poolMaxThreads;
-    const auto numWriteThreads = enoughToMultiThread ? poolMaxThreads : 1;
-    const size_t numOpsPerThread = ops.size() / numWriteThreads;
-
-    for (size_t t = 0; t < numWriteThreads; ++t) {
-        size_t begin = t * numOpsPerThread;
-        size_t end = (t == numWriteThreads - 1) ? ops.size() : begin + numOpsPerThread;
-        writerPool->schedule(makeOplogWriteForRange(begin, end));
-    }
-
-    // Wait for all scheduled writes to complete.
-    writerPool->waitForIdle();
-
-    // Reset oplogTruncateAfterPoint after writes are complete.
-    if (writeOplogColl) {
-        _consistencyMarkers->setOplogTruncateAfterPoint(opCtx, Timestamp());
-    }
-
-    return true;
-}
-
-template <typename T>
-void OplogWriterImpl::_writeOplogBatchForRange(OperationContext* opCtx,
-                                               const std::vector<T>& ops,
-                                               size_t begin,
-                                               size_t end,
-                                               bool writeOplogColl,
-                                               bool writeChangeColl) {
+void OplogWriterImpl::_writeOplogBatchImpl(OperationContext* opCtx,
+                                           const std::vector<InsertStatement>& docs,
+                                           bool writeOplogColl,
+                                           bool writeChangeColl) {
     // Oplog writes are crucial to the stability of the replica set. We give the operations
     // Immediate priority so that it skips waiting for ticket acquisition and flow control.
     ScopedAdmissionPriority<ExecutionAdmissionContext> priority(
         opCtx, AdmissionContext::Priority::kExempt);
     UnreplicatedWritesBlock uwb(opCtx);
 
-    auto docs = createInsertStatements(ops, begin, end);
-
     // The 'nsOrUUID' is used only to log the debug message when retrying inserts on the
     // oplog and change collections. The 'writeConflictRetry' helper assumes operations
     // are done on a single a single namespace. But the provided insert function can do
diff --git a/src/mongo/db/repl/oplog_writer_impl.h b/src/mongo/db/repl/oplog_writer_impl.h
index 78fe7f015db..3a4ba11de5e 100644
--- a/src/mongo/db/repl/oplog_writer_impl.h
+++ b/src/mongo/db/repl/oplog_writer_impl.h
@@ -30,9 +30,9 @@
 #pragma once
 
 #include "mongo/db/repl/oplog_writer.h"
-#include "mongo/db/repl/replication_consistency_markers.h"
 #include "mongo/db/repl/storage_interface.h"
 #include "mongo/db/stats/timer_stats.h"
+#include "mongo/util/concurrency/thread_pool.h"
 
 namespace mongo {
 namespace repl {
@@ -75,31 +75,24 @@ public:
                     OplogBuffer* applyBuffer,
                     ReplicationCoordinator* replCoord,
                     StorageInterface* storageInterface,
-                    ReplicationConsistencyMarkers* consistencyMarkers,
                     Observer* observer,
                     const OplogWriter::Options& options);
 
     /**
-     * Writes a batch of oplog entries to the oplog and/or the change collections.
+     * Writes a batch of oplog entries to the oplog and/or the change collection.
      *
-     * Returns false if nothing is written, true otherwise.
+     * The current implementation uses one thread to write to the oplog collection,
+     * and in serverless environment uses another thread to write to the serverless
+     * change collection in parallel.
      *
-     * If 'writerPool' is not set, the caller thread is used to perform the writes,
-     * otherwise 'writerPool' is used to perform the writes with multiple threads.
+     * If the batch write is successful, returns the optime of the last op written,
+     * which should be the last op in the batch.
      *
      * External states such as oplog visibility, replication opTimes and journaling
      * are not updated in this function.
      */
-    bool writeOplogBatch(OperationContext* opCtx,
-                         const std::vector<BSONObj>& ops,
-                         ThreadPool* writerPool = nullptr) override;
-
-    /**
-     * Same as above, except for the type of the oplog entries.
-     */
-    bool writeOplogBatch(OperationContext* opCtx,
-                         const std::vector<OplogEntry>& ops,
-                         ThreadPool* writerPool = nullptr) override;
+    StatusWith<OpTime> writeOplogBatch(OperationContext* opCtx,
+                                       const std::vector<BSONObj>& ops) override;
 
     /**
      * Finalizes the batch after writing it to storage, which updates various external
@@ -122,30 +115,18 @@ private:
      */
     void _run() override;
 
-    template <typename T>
-    bool _writeOplogBatch(OperationContext* opCtx,
-                          const std::vector<T>& ops,
-                          ThreadPool* writerPool);
-
-    template <typename T>
-    void _writeOplogBatchForRange(OperationContext* opCtx,
-                                  const std::vector<T>& ops,
-                                  size_t begin,
-                                  size_t end,
-                                  bool writeOplogColl,
-                                  bool writeChangeColl);
+    void _writeOplogBatchImpl(OperationContext* opCtx,
+                              const std::vector<InsertStatement>& docs,
+                              bool writeOplogColl,
+                              bool writeChangeColl);
 
-    // Not owned by us.
     OplogBuffer* const _applyBuffer;
 
     // Not owned by us.
     ReplicationCoordinator* const _replCoord;
 
     // Not owned by us.
-    StorageInterface* const _storageInterface;
-
-    // Not owned by us.
-    ReplicationConsistencyMarkers* const _consistencyMarkers;
+    StorageInterface* _storageInterface;
 
     // Not owned by us.
     Observer* const _observer;
diff --git a/src/mongo/db/repl/oplog_writer_impl_test.cpp b/src/mongo/db/repl/oplog_writer_impl_test.cpp
index 587999c34d3..7e52fed7bcd 100644
--- a/src/mongo/db/repl/oplog_writer_impl_test.cpp
+++ b/src/mongo/db/repl/oplog_writer_impl_test.cpp
@@ -33,7 +33,6 @@
 #include "mongo/db/repl/oplog_batcher_test_fixture.h"
 #include "mongo/db/repl/oplog_writer.h"
 #include "mongo/db/repl/oplog_writer_impl.h"
-#include "mongo/db/repl/replication_consistency_markers_mock.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/repl/storage_interface.h"
 #include "mongo/db/repl/storage_interface_impl.h"
@@ -105,16 +104,12 @@ protected:
 
     OperationContext* opCtx() const;
 
-    ThreadPool* getWriterPool() const;
     ReplicationCoordinator* getReplCoord() const;
     StorageInterface* getStorageInterface() const;
-    ReplicationConsistencyMarkers* getConsistencyMarkers() const;
     JournalListenerMock* getJournalListener() const;
 
     ServiceContext* _serviceContext;
     ServiceContext::UniqueOperationContext _opCtxHolder;
-    std::unique_ptr<ThreadPool> _writerPool;
-    std::unique_ptr<ReplicationConsistencyMarkers> _consistencyMarkers;
     std::unique_ptr<CountOpsObserver> _observer;
 };
 
@@ -130,8 +125,6 @@ void OplogWriterImplTest::setUp() {
 
     StorageInterface::set(_serviceContext, std::make_unique<StorageInterfaceImpl>());
 
-    _consistencyMarkers = std::make_unique<ReplicationConsistencyMarkersMock>();
-
     MongoDSessionCatalog::set(
         _serviceContext,
         std::make_unique<MongoDSessionCatalog>(
@@ -139,14 +132,11 @@ void OplogWriterImplTest::setUp() {
 
     repl::createOplog(opCtx());
 
-    _writerPool = makeReplWriterPool();
     _observer = std::make_unique<CountOpsObserver>();
 }
 
 void OplogWriterImplTest::tearDown() {
     _opCtxHolder = {};
-    _writerPool = {};
-    _consistencyMarkers = {};
     _observer = {};
     StorageInterface::set(_serviceContext, {});
     ServiceContextMongoDTest::tearDown();
@@ -156,10 +146,6 @@ OperationContext* OplogWriterImplTest::opCtx() const {
     return _opCtxHolder.get();
 }
 
-ThreadPool* OplogWriterImplTest::getWriterPool() const {
-    return _writerPool.get();
-}
-
 ReplicationCoordinator* OplogWriterImplTest::getReplCoord() const {
     return ReplicationCoordinator::get(_serviceContext);
 }
@@ -168,52 +154,43 @@ StorageInterface* OplogWriterImplTest::getStorageInterface() const {
     return StorageInterface::get(_serviceContext);
 }
 
-ReplicationConsistencyMarkers* OplogWriterImplTest::getConsistencyMarkers() const {
-    return _consistencyMarkers.get();
-}
-
 JournalListenerMock* OplogWriterImplTest::getJournalListener() const {
     return static_cast<JournalListenerMock*>(_journalListener.get());
 }
 
 DEATH_TEST_F(OplogWriterImplTest, WriteEmptyBatchFails, "!ops.empty()") {
-    OplogWriter::Options options(false /* skipWritesToOplogColl */,
-                                 false /* skipWritesToChangeColl */);
-
     OplogWriterImpl oplogWriter(nullptr,  // executor
                                 nullptr,  // writeBuffer
                                 nullptr,  // applyBuffer
                                 getReplCoord(),
                                 getStorageInterface(),
-                                getConsistencyMarkers(),
                                 &noopOplogWriterObserver,
-                                options);
+                                OplogWriter::Options());
 
     // Writing an empty batch should hit an invariant.
-    oplogWriter.writeOplogBatch(opCtx(), std::vector<BSONObj>{});
+    oplogWriter.writeOplogBatch(opCtx(), {}).getStatus().ignore();
 }
 
 TEST_F(OplogWriterImplTest, WriteOplogCollectionOnly) {
-    OplogWriter::Options options(false /* skipWritesToOplogColl */,
-                                 true /* skipWritesToChangeColl */);
-
     OplogWriterImpl oplogWriter(nullptr,  // executor
                                 nullptr,  // writeBuffer
                                 nullptr,  // applyBuffer
                                 getReplCoord(),
                                 getStorageInterface(),
-                                getConsistencyMarkers(),
                                 _observer.get(),
-                                options);
+                                OplogWriter::Options());
 
     std::vector<BSONObj> ops;
     ops.push_back(makeRawInsertOplogEntry(1, kNss1));
     ops.push_back(makeRawInsertOplogEntry(2, kNss2));
 
-    auto written = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+    auto returnOpTime = OpTime::parseFromOplogEntry(ops.back()).getValue();
+    auto statusWith = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+
+    ASSERT_OK(statusWith);
+    ASSERT_EQ(returnOpTime, statusWith.getValue());
 
     // Verify that the batch is only written to the oplog collection.
-    ASSERT(written);
     ASSERT_EQ(2, _observer->oplogCollDocsCount.load());
     ASSERT_EQ(0, _observer->changeCollDocsCount.load());
 }
@@ -226,26 +203,25 @@ TEST_F(OplogWriterImplTest, WriteChangeCollectionsOnly) {
 
     ChangeStreamChangeCollectionManager::create(_serviceContext);
 
-    OplogWriter::Options options(true /* skipWritesToOplogColl */,
-                                 false /* skipWritesToChangeColl */);
-
     OplogWriterImpl oplogWriter(nullptr,  // executor
                                 nullptr,  // writeBuffer
                                 nullptr,  // applyBuffer
                                 getReplCoord(),
                                 getStorageInterface(),
-                                getConsistencyMarkers(),
                                 _observer.get(),
-                                options);
+                                OplogWriter::Options(true /* skipWritesToOplogColl */));
 
     std::vector<BSONObj> ops;
     ops.push_back(makeRawInsertOplogEntry(1, kNss1));
     ops.push_back(makeRawInsertOplogEntry(2, kNss2));
 
-    auto written = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+    auto returnOpTime = OpTime::parseFromOplogEntry(ops.back()).getValue();
+    auto statusWith = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+
+    ASSERT_OK(statusWith);
+    ASSERT_EQ(returnOpTime, statusWith.getValue());
 
     // Verify that the batch is only written to the change collections.
-    ASSERT(written);
     ASSERT_EQ(0, _observer->oplogCollDocsCount.load());
     ASSERT_EQ(2, _observer->changeCollDocsCount.load());
 }
@@ -258,67 +234,60 @@ TEST_F(OplogWriterImplTest, WriteBothOplogAndChangeCollections) {
 
     ChangeStreamChangeCollectionManager::create(_serviceContext);
 
-    OplogWriter::Options options(false /* skipWritesToOplogColl */,
-                                 false /* skipWritesToChangeColl */);
-
     OplogWriterImpl oplogWriter(nullptr,  // executor
                                 nullptr,  // writeBuffer
                                 nullptr,  // applyBuffer
                                 getReplCoord(),
                                 getStorageInterface(),
-                                getConsistencyMarkers(),
                                 _observer.get(),
-                                options);
+                                OplogWriter::Options());
 
     std::vector<BSONObj> ops;
     ops.push_back(makeRawInsertOplogEntry(1, kNss1));
     ops.push_back(makeRawInsertOplogEntry(2, kNss2));
 
-    auto written = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+    auto returnOpTime = OpTime::parseFromOplogEntry(ops.back()).getValue();
+    auto statusWith = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+
+    ASSERT_OK(statusWith);
+    ASSERT_EQ(returnOpTime, statusWith.getValue());
 
     // Verify that the batch written to both the oplog and change collections.
-    ASSERT(written);
     ASSERT_EQ(2, _observer->oplogCollDocsCount.load());
     ASSERT_EQ(2, _observer->changeCollDocsCount.load());
 }
 
 TEST_F(OplogWriterImplTest, WriteNeitherOplogNorChangeCollections) {
-    OplogWriter::Options options(true /* skipWritesToOplogColl */,
-                                 true /* skipWritesToChangeColl */);
-
     OplogWriterImpl oplogWriter(nullptr,  // executor
                                 nullptr,  // writeBuffer
                                 nullptr,  // applyBuffer
                                 getReplCoord(),
                                 getStorageInterface(),
-                                getConsistencyMarkers(),
                                 _observer.get(),
-                                options);
+                                OplogWriter::Options(true /* skipWritesToOplogColl */));
 
     std::vector<BSONObj> ops;
     ops.push_back(makeRawInsertOplogEntry(1, kNss1));
     ops.push_back(makeRawInsertOplogEntry(2, kNss2));
 
-    auto written = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+    auto statusWith = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+
+    ASSERT_OK(statusWith);
+    ASSERT_EQ(OpTime(), statusWith.getValue());
 
-    // Verify that the batch is not written any collection.
-    ASSERT(!written);
+    // Verify that the batch is only written to the oplog collection.
     ASSERT_EQ(0, _observer->oplogCollDocsCount.load());
     ASSERT_EQ(0, _observer->changeCollDocsCount.load());
 }
 
 TEST_F(OplogWriterImplTest, finalizeOplogBatchCorrectlyUpdatesOpTimes) {
-    OplogWriter::Options options(false /* skipWritesToOplogColl */,
-                                 false /* skipWritesToChangeColl */);
-
     OplogWriterImpl oplogWriter(nullptr,  // executor
                                 nullptr,  // writeBuffer
                                 nullptr,  // applyBuffer
                                 getReplCoord(),
                                 getStorageInterface(),
-                                getConsistencyMarkers(),
                                 &noopOplogWriterObserver,
-                                options);
+                                OplogWriter::Options());
 
     auto curOpTime = OpTime(Timestamp(2, 2), 1);
     auto curWallTime = Date_t::now();
diff --git a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
index 1e1cc64f850..21d1189e0ea 100644
--- a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
@@ -297,28 +297,29 @@ void ReplicationCoordinatorExternalStateImpl::startSteadyStateReplication(
 
     // Using noop observer for both writer and applier. During steady state replication,
     // there is no need to log details on every batch we apply.
+    // TODO (SERVER-87674): use a different thread pool.
     if (useOplogWriter) {
-        _oplogWriter = std::make_unique<OplogWriterImpl>(
-            _oplogWriterTaskExecutor.get(),
-            _oplogWriteBuffer.get(),
-            _oplogApplyBuffer.get(),
-            replCoord,
-            _storageInterface,
-            _replicationProcess->getConsistencyMarkers(),
-            &noopOplogWriterObserver,
-            OplogWriter::Options(false /* skipWritesToOplogColl */,
-                                 true /* skipWritesToChangeColl */));
+        _oplogWriter = std::make_unique<OplogWriterImpl>(_oplogWriterTaskExecutor.get(),
+                                                         _oplogWriteBuffer.get(),
+                                                         _oplogApplyBuffer.get(),
+                                                         replCoord,
+                                                         _storageInterface,
+                                                         &noopOplogWriterObserver,
+                                                         OplogWriter::Options());
     }
 
-    _oplogApplier = std::make_unique<OplogApplierImpl>(
-        _oplogApplierTaskExecutor.get(),
-        _oplogApplyBuffer.get(),
-        &noopOplogApplierObserver,
-        replCoord,
-        _replicationProcess->getConsistencyMarkers(),
-        _storageInterface,
-        OplogApplier::Options(OplogApplication::Mode::kSecondary),
-        _writerPool.get());
+    // TODO (SERVER-85697): clean up the applier options.
+    OplogApplier::Options applierOptions(OplogApplication::Mode::kSecondary,
+                                         useOplogWriter /* skipWritesToOplog */,
+                                         useOplogWriter /* skipWritesToChangeCollection */);
+    _oplogApplier = std::make_unique<OplogApplierImpl>(_oplogApplierTaskExecutor.get(),
+                                                       _oplogApplyBuffer.get(),
+                                                       &noopOplogApplierObserver,
+                                                       replCoord,
+                                                       _replicationProcess->getConsistencyMarkers(),
+                                                       _storageInterface,
+                                                       applierOptions,
+                                                       _writerPool.get());
 
     invariant(!_bgSync);
     _bgSync = std::make_unique<BackgroundSync>(