SERVER-72502 Implement OplogBatcher changes for parallelizing prepared transaction oplog application.

1 files changed, 5 insertions, 11 deletions

diff --git a/src/mongo/db/repl/oplog_applier_impl.cpp b/src/mongo/db/repl/oplog_applier_impl.cpp
index 1340eb53b64..baff24f4ba7 100644
--- a/src/mongo/db/repl/oplog_applier_impl.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl.cpp
@@ -808,13 +808,6 @@ void OplogApplierImpl::_deriveOpsAndFillWriterVectors(
             continue;
         }
 
-        // Clear the partialTxnList if the transaction needs to be aborted.
-        if (op.getCommandType() == OplogEntry::CommandType::kAbortTransaction) {
-            // Under current oplog batching rules, it is not possible for abortTransaction to
-            // be in the same batch as a preceding partial transaction applyOps oplog entry.
-            invariant(partialTxnOps.empty(), op.toStringForLogging());
-        }
-
         // Extract applyOps operations and fill writers with extracted operations.
         if (op.isTerminalApplyOps()) {
             // applyOps entries generated by a transaction must have a prevOpTime.
@@ -855,7 +848,7 @@ void OplogApplierImpl::_deriveOpsAndFillWriterVectors(
 
             // Fill the writers with commit or abort operation. Depending on whether the operation
             // refers to a split prepare, it might also be split into multiple ops.
-            if ((op.isPreparedCommit() || op.isPreparedAbort()) &&
+            if (op.isPreparedCommitOrAbort() &&
                 (getOptions().mode == OplogApplication::Mode::kSecondary)) {
                 OplogApplierUtils::addDerivedCommitsOrAborts(
                     opCtx, &op, writerVectors, &collPropertiesCache);
@@ -906,9 +899,10 @@ Status applyOplogEntryOrGroupedInserts(OperationContext* opCtx,
                                        const OplogEntryOrGroupedInserts& entryOrGroupedInserts,
                                        OplogApplication::Mode oplogApplicationMode,
                                        const bool isDataConsistent) {
-    // Guarantees that applyOplogEntryOrGroupedInserts' context matches that of its calling
-    // function, applyOplogBatchPerWorker.
-    invariant(!opCtx->writesAreReplicated());
+    // Certain operations like prepareTransaction might reset the recovery unit or lock state
+    // due to doing things like stashTransactionResources. So we restore the necessary states
+    // here every time before applying a new entry or grouped inserts.
+    _setOplogApplicationWorkerOpCtxStates(opCtx);
 
     // Count each log op application as a separate operation, for reporting purposes