diff --git a/buildSrc/src/main/resources/checkstyle_suppressions.xml b/buildSrc/src/main/resources/checkstyle_suppressions.xml
index 8d0fbe3edda..c95ad03f9ac 100644
--- a/buildSrc/src/main/resources/checkstyle_suppressions.xml
+++ b/buildSrc/src/main/resources/checkstyle_suppressions.xml
@@ -2304,7 +2304,6 @@
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]RetryTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]TransportBulkActionIngestTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]TransportBulkActionTookTests.java" checks="LineLength" />
-  <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]TransportShardBulkActionTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]byscroll[/\\]AsyncBulkByScrollActionTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]byscroll[/\\]BulkByScrollParallelizationHelperTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]byscroll[/\\]BulkByScrollResponseTests.java" checks="LineLength" />
diff --git a/core/src/main/java/org/elasticsearch/action/bulk/MappingUpdatePerformer.java b/core/src/main/java/org/elasticsearch/action/bulk/MappingUpdatePerformer.java
index 27d636d3d93..e33c54d6dd5 100644
--- a/core/src/main/java/org/elasticsearch/action/bulk/MappingUpdatePerformer.java
+++ b/core/src/main/java/org/elasticsearch/action/bulk/MappingUpdatePerformer.java
@@ -20,48 +20,28 @@
 package org.elasticsearch.action.bulk;
 
 import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.shard.IndexShard;
+import org.elasticsearch.index.shard.ShardId;
 
 import java.util.Objects;
 
 public interface MappingUpdatePerformer {
-    /**
-     * Determine if any mappings need to be updated, and update them on the
-     * master node if necessary. Returnes a failed {@code Engine.IndexResult}
-     * in the event updating the mappings fails or null if successful.
-     * Throws a {@code ReplicationOperation.RetryOnPrimaryException} if the
-     * operation needs to be retried on the primary due to the mappings not
-     * being present yet, or a different exception if updating the mappings
-     * on the master failed.
-     */
-    @Nullable
-    MappingUpdateResult updateMappingsIfNeeded(IndexShard primary, IndexRequest request) throws Exception;
 
     /**
-     * Class encapsulating the resulting of potentially updating the mapping
+     * Determine if any mappings need to be updated, and update them on the master node if
+     * necessary. Returnes a failure Exception in the event updating the mappings fails or null if
+     * successful.
      */
-    class MappingUpdateResult {
-        @Nullable
-        public final Engine.Index operation;
-        @Nullable
-        public final Exception failure;
+    void updateMappingsIfNeeded(Engine.Index operation,
+                                ShardId shardId,
+                                String type) throws Exception;
 
-        MappingUpdateResult(Exception failure) {
-            Objects.requireNonNull(failure, "failure cannot be null");
-            this.failure = failure;
-            this.operation = null;
-        }
+    /**
+     *  Throws a {@code ReplicationOperation.RetryOnPrimaryException} if the operation needs to be
+     * retried on the primary due to the mappings not being present yet, or a different exception if
+     * updating the mappings on the master failed.
+     */
+    void verifyMappings(Engine.Index operation, ShardId shardId) throws Exception;
 
-        MappingUpdateResult(Engine.Index operation) {
-            Objects.requireNonNull(operation, "operation cannot be null");
-            this.operation = operation;
-            this.failure = null;
-        }
-
-        public boolean isFailed() {
-            return failure != null;
-        }
-    }
 }
diff --git a/core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java b/core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
index d1bba8d2d4a..6a286c5a758 100644
--- a/core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
+++ b/core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
@@ -65,6 +65,9 @@ import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportRequestOptions;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.index.translog.Translog.Location;
+import org.elasticsearch.action.bulk.BulkItemResultHolder;
+import org.elasticsearch.action.bulk.BulkItemResponse;
 
 import java.io.IOException;
 import java.util.Map;
@@ -154,10 +157,23 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
         }
     }
 
+    static Translog.Location calculateTranslogLocation(final Translog.Location originalLocation,
+                                                       final BulkItemResultHolder bulkItemResult) {
+        final Engine.Result operationResult = bulkItemResult.operationResult;
+        if (operationResult != null && operationResult.hasFailure() == false) {
+            return locationToSync(originalLocation, operationResult.getTranslogLocation());
+        } else {
+            return originalLocation;
+        }
+    }
+
     // Visible for unit testing
-    static Translog.Location updateReplicaRequest(BulkItemResultHolder bulkItemResult,
+    /**
+     * Creates a BulkItemResponse for the primary operation and returns it. If no bulk response is
+     * needed (because one already exists and the operation failed), then return null.
+     */
+    static BulkItemResponse createPrimaryResponse(BulkItemResultHolder bulkItemResult,
                                                   final DocWriteRequest.OpType opType,
-                                                  final Translog.Location originalLocation,
                                                   BulkShardRequest request) {
         final Engine.Result operationResult = bulkItemResult.operationResult;
         final DocWriteResponse response = bulkItemResult.response;
@@ -165,16 +181,13 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
 
         if (operationResult == null) { // in case of noop update operation
             assert response.getResult() == DocWriteResponse.Result.NOOP : "only noop updates can have a null operation";
-            replicaRequest.setPrimaryResponse(new BulkItemResponse(replicaRequest.id(), opType, response));
-            return originalLocation;
+            return new BulkItemResponse(replicaRequest.id(), opType, response);
 
         } else if (operationResult.hasFailure() == false) {
             BulkItemResponse primaryResponse = new BulkItemResponse(replicaRequest.id(), opType, response);
-            replicaRequest.setPrimaryResponse(primaryResponse);
             // set a blank ShardInfo so we can safely send it to the replicas. We won't use it in the real response though.
             primaryResponse.getResponse().setShardInfo(new ShardInfo());
-            // The operation was successful, advance the translog
-            return locationToSync(originalLocation, operationResult.getTranslogLocation());
+            return primaryResponse;
 
         } else {
             DocWriteRequest docWriteRequest = replicaRequest.request();
@@ -187,19 +200,19 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
                     request.shardId(), docWriteRequest.opType().getLowercase(), request), failure);
             }
 
-
             // if it's a conflict failure, and we already executed the request on a primary (and we execute it
             // again, due to primary relocation and only processing up to N bulk items when the shard gets closed)
             // then just use the response we got from the failed execution
             if (replicaRequest.getPrimaryResponse() == null || isConflictException(failure) == false) {
-                replicaRequest.setPrimaryResponse(
-                        new BulkItemResponse(replicaRequest.id(), docWriteRequest.opType(),
-                                // Make sure to use request.indox() here, if you
-                                // use docWriteRequest.index() it will use the
-                                // concrete index instead of an alias if used!
-                                new BulkItemResponse.Failure(request.index(), docWriteRequest.type(), docWriteRequest.id(), failure)));
+                return new BulkItemResponse(replicaRequest.id(), docWriteRequest.opType(),
+                        // Make sure to use request.index() here, if you
+                        // use docWriteRequest.index() it will use the
+                        // concrete index instead of an alias if used!
+                        new BulkItemResponse.Failure(request.index(), docWriteRequest.type(), docWriteRequest.id(), failure));
+            } else {
+                assert replicaRequest.getPrimaryResponse() != null : "replica request must have a primary response";
+                return null;
             }
-            return originalLocation;
         }
     }
 
@@ -233,11 +246,14 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
         // update the bulk item request because update request execution can mutate the bulk item request
         request.items()[requestIndex] = replicaRequest;
 
-        // Modify the replica request, if needed, and return a new translog location
-        location = updateReplicaRequest(responseHolder, opType, location, request);
+        // Retrieve the primary response, and update the replica request with the primary's response
+        BulkItemResponse primaryResponse = createPrimaryResponse(responseHolder, opType, request);
+        if (primaryResponse != null) {
+            replicaRequest.setPrimaryResponse(primaryResponse);
+        }
 
-        assert replicaRequest.getPrimaryResponse() != null : "replica request must have a primary response";
-        return location;
+        // Update the translog with the new location, if needed
+        return calculateTranslogLocation(location, responseHolder);
     }
 
     private static boolean isConflictException(final Exception e) {
@@ -396,14 +412,16 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
         return new WriteReplicaResult<>(request, location, null, replica, logger);
     }
 
-    private static Translog.Location locationToSync(Translog.Location current, Translog.Location next) {
-        /* here we are moving forward in the translog with each operation. Under the hood
-         * this might cross translog files which is ok since from the user perspective
-         * the translog is like a tape where only the highest location needs to be fsynced
-         * in order to sync all previous locations even though they are not in the same file.
-         * When the translog rolls over files the previous file is fsynced on after closing if needed.*/
+    private static Translog.Location locationToSync(Translog.Location current,
+                                                    Translog.Location next) {
+        /* here we are moving forward in the translog with each operation. Under the hood this might
+         * cross translog files which is ok since from the user perspective the translog is like a
+         * tape where only the highest location needs to be fsynced in order to sync all previous
+         * locations even though they are not in the same file. When the translog rolls over files
+         * the previous file is fsynced on after closing if needed.*/
         assert next != null : "next operation can't be null";
-        assert current == null || current.compareTo(next) < 0 : "translog locations are not increasing";
+        assert current == null || current.compareTo(next) < 0 :
+                "translog locations are not increasing";
         return next;
     }
 
@@ -411,45 +429,82 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
      * Execute the given {@link IndexRequest} on a replica shard, throwing a
      * {@link RetryOnReplicaException} if the operation needs to be re-tried.
      */
-    public static Engine.IndexResult executeIndexRequestOnReplica(DocWriteResponse primaryResponse, IndexRequest request, IndexShard replica) throws IOException {
-        final ShardId shardId = replica.shardId();
-        SourceToParse sourceToParse =
-            SourceToParse.source(SourceToParse.Origin.REPLICA, shardId.getIndexName(), request.type(), request.id(), request.source(),
-                request.getContentType()).routing(request.routing()).parent(request.parent());
+    public static Engine.IndexResult executeIndexRequestOnReplica(
+            DocWriteResponse primaryResponse,
+            IndexRequest request,
+            IndexShard replica) throws IOException {
 
         final Engine.Index operation;
-        final long version = primaryResponse.getVersion();
-        final VersionType versionType = request.versionType().versionTypeForReplicationAndRecovery();
-        assert versionType.validateVersionForWrites(version);
-        final long seqNo = primaryResponse.getSeqNo();
         try {
-            operation = replica.prepareIndexOnReplica(sourceToParse, seqNo, version, versionType, request.getAutoGeneratedTimestamp(), request.isRetry());
+            operation = prepareIndexOperationOnReplica(primaryResponse, request, replica);
         } catch (MapperParsingException e) {
-            return new Engine.IndexResult(e, version, seqNo);
+            return new Engine.IndexResult(e, primaryResponse.getVersion(),
+                    primaryResponse.getSeqNo());
         }
+
         Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
         if (update != null) {
-            throw new RetryOnReplicaException(shardId, "Mappings are not available on the replica yet, triggered update: " + update);
+            final ShardId shardId = replica.shardId();
+            throw new RetryOnReplicaException(shardId,
+                    "Mappings are not available on the replica yet, triggered update: " + update);
         }
         return replica.index(operation);
     }
 
+    /** Utility method to prepare an index operation on replica shards */
+    static Engine.Index prepareIndexOperationOnReplica(
+            DocWriteResponse primaryResponse,
+            IndexRequest request,
+            IndexShard replica) {
+
+        final ShardId shardId = replica.shardId();
+        final long version = primaryResponse.getVersion();
+        final long seqNo = primaryResponse.getSeqNo();
+        final SourceToParse sourceToParse =
+                SourceToParse.source(SourceToParse.Origin.REPLICA, shardId.getIndexName(),
+                        request.type(), request.id(), request.source(), request.getContentType())
+                .routing(request.routing()).parent(request.parent());
+        final VersionType versionType = request.versionType().versionTypeForReplicationAndRecovery();
+        assert versionType.validateVersionForWrites(version);
+
+        return replica.prepareIndexOnReplica(sourceToParse, seqNo, version, versionType,
+                request.getAutoGeneratedTimestamp(), request.isRetry());
+    }
+
     /** Utility method to prepare an index operation on primary shards */
     static Engine.Index prepareIndexOperationOnPrimary(IndexRequest request, IndexShard primary) {
-        SourceToParse sourceToParse =
-            SourceToParse.source(SourceToParse.Origin.PRIMARY, request.index(), request.type(), request.id(), request.source(),
-                request.getContentType()).routing(request.routing()).parent(request.parent());
-        return primary.prepareIndexOnPrimary(sourceToParse, request.version(), request.versionType(), request.getAutoGeneratedTimestamp(), request.isRetry());
+        final SourceToParse sourceToParse =
+                SourceToParse.source(SourceToParse.Origin.PRIMARY, request.index(), request.type(),
+                        request.id(), request.source(), request.getContentType())
+                .routing(request.routing()).parent(request.parent());
+        return primary.prepareIndexOnPrimary(sourceToParse, request.version(), request.versionType(),
+                request.getAutoGeneratedTimestamp(), request.isRetry());
     }
 
     /** Executes index operation on primary shard after updates mapping if dynamic mappings are found */
     public static Engine.IndexResult executeIndexRequestOnPrimary(IndexRequest request, IndexShard primary,
                                                                   MappingUpdatePerformer mappingUpdater) throws Exception {
-        MappingUpdatePerformer.MappingUpdateResult result = mappingUpdater.updateMappingsIfNeeded(primary, request);
-        if (result.isFailed()) {
-            return new Engine.IndexResult(result.failure, request.version());
+        // Update the mappings if parsing the documents includes new dynamic updates
+        try {
+            final Engine.Index preUpdateOperation = prepareIndexOperationOnPrimary(request, primary);
+            mappingUpdater.updateMappingsIfNeeded(preUpdateOperation, primary.shardId(), request.type());
+        } catch (MapperParsingException | IllegalArgumentException failure) {
+            return new Engine.IndexResult(failure, request.version());
         }
-        return primary.index(result.operation);
+
+        // Verify that there are no more mappings that need to be applied. If there are failures, a
+        // ReplicationOperation.RetryOnPrimaryException is thrown.
+        final Engine.Index operation;
+        try {
+            operation = prepareIndexOperationOnPrimary(request, primary);
+            mappingUpdater.verifyMappings(operation, primary.shardId());
+        } catch (MapperParsingException | IllegalStateException e) {
+            // there was an error in parsing the document that was not because
+            // of pending mapping updates, so return a failure for the result
+            return new Engine.IndexResult(e, request.version());
+        }
+
+        return primary.index(operation);
     }
 
     private static Engine.DeleteResult executeDeleteRequestOnPrimary(DeleteRequest request, IndexShard primary) throws IOException {
@@ -468,36 +523,22 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
 
     class ConcreteMappingUpdatePerformer implements MappingUpdatePerformer {
 
-        @Nullable
-        public MappingUpdateResult updateMappingsIfNeeded(IndexShard primary, IndexRequest request) throws Exception {
-            Engine.Index operation;
-            try {
-                operation = prepareIndexOperationOnPrimary(request, primary);
-            } catch (MapperParsingException | IllegalArgumentException e) {
-                return new MappingUpdateResult(e);
-            }
+        public void updateMappingsIfNeeded(final Engine.Index operation, final ShardId shardId,
+                                           final String type) throws Exception {
             final Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
-            final ShardId shardId = primary.shardId();
             if (update != null) {
-                // can throw timeout exception when updating mappings or ISE for attempting to update default mappings
-                // which are bubbled up
-                try {
-                    mappingUpdatedAction.updateMappingOnMaster(shardId.getIndex(), request.type(), update);
-                } catch (IllegalArgumentException e) {
-                    // throws IAE on conflicts merging dynamic mappings
-                    return new MappingUpdateResult(e);
-                }
-                try {
-                    operation = prepareIndexOperationOnPrimary(request, primary);
-                } catch (MapperParsingException | IllegalArgumentException e) {
-                    return new MappingUpdateResult(e);
-                }
-                if (operation.parsedDoc().dynamicMappingsUpdate() != null) {
-                    throw new ReplicationOperation.RetryOnPrimaryException(shardId,
-                            "Dynamic mappings are not available on the node that holds the primary yet");
-                }
+                // can throw timeout exception when updating mappings or ISE for attempting to
+                // update default mappings which are bubbled up
+                mappingUpdatedAction.updateMappingOnMaster(shardId.getIndex(), type, update);
+            }
+        }
+
+        public void verifyMappings(final Engine.Index operation,
+                                   final ShardId shardId) throws Exception {
+            if (operation.parsedDoc().dynamicMappingsUpdate() != null) {
+                throw new ReplicationOperation.RetryOnPrimaryException(shardId,
+                        "Dynamic mappings are not available on the node that holds the primary yet");
             }
-            return new MappingUpdateResult(operation);
         }
     }
 }
diff --git a/core/src/test/java/org/elasticsearch/action/bulk/TransportShardBulkActionTests.java b/core/src/test/java/org/elasticsearch/action/bulk/TransportShardBulkActionTests.java
index c30c2ac2f13..a11dea344ed 100644
--- a/core/src/test/java/org/elasticsearch/action/bulk/TransportShardBulkActionTests.java
+++ b/core/src/test/java/org/elasticsearch/action/bulk/TransportShardBulkActionTests.java
@@ -53,6 +53,8 @@ import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.action.bulk.TransportShardBulkAction;
+import org.elasticsearch.action.bulk.MappingUpdatePerformer;
+import org.elasticsearch.action.bulk.BulkItemResultHolder;
 
 import java.io.IOException;
 import static org.hamcrest.CoreMatchers.equalTo;
@@ -77,26 +79,30 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
 
     public void testShouldExecuteReplicaItem() throws Exception {
         // Successful index request should be replicated
-        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id").source(Requests.INDEX_CONTENT_TYPE, "foo", "bar");
+        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id")
+                .source(Requests.INDEX_CONTENT_TYPE, "foo", "bar");
         DocWriteResponse response = new IndexResponse(shardId, "type", "id", 1, 1, randomBoolean());
         BulkItemRequest request = new BulkItemRequest(0, writeRequest);
         request.setPrimaryResponse(new BulkItemResponse(0, DocWriteRequest.OpType.INDEX, response));
         assertTrue(TransportShardBulkAction.shouldExecuteReplicaItem(request, 0));
 
         // Failed index requests should not be replicated (for now!)
-        writeRequest = new IndexRequest("index", "type", "id").source(Requests.INDEX_CONTENT_TYPE, "foo", "bar");
+        writeRequest = new IndexRequest("index", "type", "id")
+                .source(Requests.INDEX_CONTENT_TYPE, "foo", "bar");
         response = new IndexResponse(shardId, "type", "id", 1, 1, randomBoolean());
         request = new BulkItemRequest(0, writeRequest);
         request.setPrimaryResponse(
                 new BulkItemResponse(0, DocWriteRequest.OpType.INDEX,
-                        new BulkItemResponse.Failure("test", "type", "id", new IllegalArgumentException("i died"))));
+                        new BulkItemResponse.Failure("test", "type", "id",
+                                                     new IllegalArgumentException("i died"))));
         assertFalse(TransportShardBulkAction.shouldExecuteReplicaItem(request, 0));
 
         // NOOP requests should not be replicated
         writeRequest = new UpdateRequest("index", "type", "id");
         response = new UpdateResponse(shardId, "type", "id", 1, DocWriteResponse.Result.NOOP);
         request = new BulkItemRequest(0, writeRequest);
-        request.setPrimaryResponse(new BulkItemResponse(0, DocWriteRequest.OpType.UPDATE, response));
+        request.setPrimaryResponse(new BulkItemResponse(0, DocWriteRequest.OpType.UPDATE,
+                        response));
         assertFalse(TransportShardBulkAction.shouldExecuteReplicaItem(request, 0));
     }
 
@@ -112,13 +118,15 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
                 .create(create);
         BulkItemRequest primaryRequest = new BulkItemRequest(0, writeRequest);
         items[0] = primaryRequest;
-        BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
+        BulkShardRequest bulkShardRequest =
+                new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
 
         Translog.Location location = new Translog.Location(0, 0, 0);
         UpdateHelper updateHelper = null;
 
-        Translog.Location newLocation = TransportShardBulkAction.executeBulkItemRequest(metaData, shard, bulkShardRequest,
-                location, 0, updateHelper, threadPool::absoluteTimeInMillis, new NoopMappingUpdatePerformer());
+        Translog.Location newLocation = TransportShardBulkAction.executeBulkItemRequest(metaData,
+                shard, bulkShardRequest, location, 0, updateHelper,
+                threadPool::absoluteTimeInMillis, new NoopMappingUpdatePerformer());
 
         // Translog should change, since there were no problems
         assertThat(newLocation, not(location));
@@ -127,7 +135,8 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
 
         assertThat(primaryResponse.getItemId(), equalTo(0));
         assertThat(primaryResponse.getId(), equalTo("id"));
-        assertThat(primaryResponse.getOpType(), equalTo(create ? DocWriteRequest.OpType.CREATE : DocWriteRequest.OpType.INDEX));
+        assertThat(primaryResponse.getOpType(),
+                equalTo(create ? DocWriteRequest.OpType.CREATE : DocWriteRequest.OpType.INDEX));
         assertFalse(primaryResponse.isFailed());
 
         // Assert that the document actually made it there
@@ -140,8 +149,10 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
         items[0] = primaryRequest;
         bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
 
-        Translog.Location secondLocation = TransportShardBulkAction.executeBulkItemRequest(metaData, shard, bulkShardRequest,
-                newLocation, 0, updateHelper, threadPool::absoluteTimeInMillis, new NoopMappingUpdatePerformer());
+        Translog.Location secondLocation =
+                TransportShardBulkAction.executeBulkItemRequest( metaData,
+                        shard, bulkShardRequest, newLocation, 0, updateHelper,
+                        threadPool::absoluteTimeInMillis, new NoopMappingUpdatePerformer());
 
         // Translog should not change, since the document was not indexed due to a version conflict
         assertThat(secondLocation, equalTo(newLocation));
@@ -177,9 +188,11 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
         IndexShard shard = newStartedShard(true);
 
         BulkItemRequest[] items = new BulkItemRequest[1];
-        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id").source(Requests.INDEX_CONTENT_TYPE, "foo", "bar");
+        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id")
+                .source(Requests.INDEX_CONTENT_TYPE, "foo", "bar");
         items[0] = new BulkItemRequest(0, writeRequest);
-        BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
+        BulkShardRequest bulkShardRequest =
+                new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
 
         Translog.Location location = new Translog.Location(0, 0, 0);
         UpdateHelper updateHelper = null;
@@ -188,8 +201,9 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
         Exception err = new ReplicationOperation.RetryOnPrimaryException(shardId, "rejection");
 
         try {
-            TransportShardBulkAction.executeBulkItemRequest(metaData, shard, bulkShardRequest, location,
-                    0, updateHelper, threadPool::absoluteTimeInMillis, new ThrowingMappingUpdatePerformer(err));
+            TransportShardBulkAction.executeBulkItemRequest(metaData, shard, bulkShardRequest,
+                    location, 0, updateHelper, threadPool::absoluteTimeInMillis,
+                    new ThrowingMappingUpdatePerformer(err));
             fail("should have thrown a retry exception");
         } catch (ReplicationOperation.RetryOnPrimaryException e) {
             assertThat(e, equalTo(err));
@@ -203,9 +217,11 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
         IndexShard shard = newStartedShard(true);
 
         BulkItemRequest[] items = new BulkItemRequest[1];
-        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id").source(Requests.INDEX_CONTENT_TYPE, "foo", "bar");
+        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id")
+                .source(Requests.INDEX_CONTENT_TYPE, "foo", "bar");
         items[0] = new BulkItemRequest(0, writeRequest);
-        BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
+        BulkShardRequest bulkShardRequest =
+                new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
 
         Translog.Location location = new Translog.Location(0, 0, 0);
         UpdateHelper updateHelper = null;
@@ -213,8 +229,9 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
         // Return a mapping conflict (IAE) when trying to update the mapping
         Exception err = new IllegalArgumentException("mapping conflict");
 
-        Translog.Location newLocation = TransportShardBulkAction.executeBulkItemRequest(metaData, shard, bulkShardRequest,
-                location, 0, updateHelper, threadPool::absoluteTimeInMillis, new FailingMappingUpdatePerformer(err));
+        Translog.Location newLocation = TransportShardBulkAction.executeBulkItemRequest(metaData,
+                shard, bulkShardRequest, location, 0, updateHelper,
+                threadPool::absoluteTimeInMillis, new ThrowingMappingUpdatePerformer(err));
 
         // Translog shouldn't change, as there were conflicting mappings
         assertThat(newLocation, equalTo(location));
@@ -245,13 +262,15 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
         BulkItemRequest[] items = new BulkItemRequest[1];
         DocWriteRequest writeRequest = new DeleteRequest("index", "type", "id");
         items[0] = new BulkItemRequest(0, writeRequest);
-        BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
+        BulkShardRequest bulkShardRequest =
+                new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
 
         Translog.Location location = new Translog.Location(0, 0, 0);
         UpdateHelper updateHelper = null;
 
-        Translog.Location newLocation = TransportShardBulkAction.executeBulkItemRequest(metaData, shard, bulkShardRequest,
-                location, 0, updateHelper, threadPool::absoluteTimeInMillis, new NoopMappingUpdatePerformer());
+        Translog.Location newLocation = TransportShardBulkAction.executeBulkItemRequest(metaData,
+                shard, bulkShardRequest, location, 0, updateHelper,
+                threadPool::absoluteTimeInMillis, new NoopMappingUpdatePerformer());
 
         // Translog changes, even though the document didn't exist
         assertThat(newLocation, not(location));
@@ -288,8 +307,9 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
 
         location = newLocation;
 
-        newLocation = TransportShardBulkAction.executeBulkItemRequest(metaData, shard, bulkShardRequest,
-                location, 0, updateHelper, threadPool::absoluteTimeInMillis, new NoopMappingUpdatePerformer());
+        newLocation = TransportShardBulkAction.executeBulkItemRequest(metaData, shard,
+                bulkShardRequest, location, 0, updateHelper, threadPool::absoluteTimeInMillis,
+                new NoopMappingUpdatePerformer());
 
         // Translog changes, because the document was deleted
         assertThat(newLocation, not(location));
@@ -322,19 +342,25 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
     }
 
     public void testNoopUpdateReplicaRequest() throws Exception {
-        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id").source(Requests.INDEX_CONTENT_TYPE, "field", "value");
+        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id")
+                .source(Requests.INDEX_CONTENT_TYPE, "field", "value");
         BulkItemRequest replicaRequest = new BulkItemRequest(0, writeRequest);
 
-        DocWriteResponse noopUpdateResponse = new UpdateResponse(shardId, "index", "id", 0, DocWriteResponse.Result.NOOP);
-        BulkItemResultHolder noopResults = new BulkItemResultHolder(noopUpdateResponse, null, replicaRequest);
+        DocWriteResponse noopUpdateResponse = new UpdateResponse(shardId, "index", "id", 0,
+                DocWriteResponse.Result.NOOP);
+        BulkItemResultHolder noopResults = new BulkItemResultHolder(noopUpdateResponse, null,
+                replicaRequest);
 
         Translog.Location location = new Translog.Location(0, 0, 0);
         BulkItemRequest[] items = new BulkItemRequest[0];
-        BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
-        Translog.Location newLocation = TransportShardBulkAction.updateReplicaRequest(noopResults,
-                DocWriteRequest.OpType.UPDATE, location, bulkShardRequest);
+        BulkShardRequest bulkShardRequest =
+                new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
 
-        BulkItemResponse primaryResponse = replicaRequest.getPrimaryResponse();
+        BulkItemResponse primaryResponse = TransportShardBulkAction.createPrimaryResponse(
+                noopResults, DocWriteRequest.OpType.UPDATE, bulkShardRequest);
+
+        Translog.Location newLocation =
+                TransportShardBulkAction.calculateTranslogLocation(location, noopResults);
 
         // Basically nothing changes in the request since it's a noop
         assertThat(newLocation, equalTo(location));
@@ -342,24 +368,30 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
         assertThat(primaryResponse.getId(), equalTo("id"));
         assertThat(primaryResponse.getOpType(), equalTo(DocWriteRequest.OpType.UPDATE));
         assertThat(primaryResponse.getResponse(), equalTo(noopUpdateResponse));
-        assertThat(primaryResponse.getResponse().getResult(), equalTo(DocWriteResponse.Result.NOOP));
+        assertThat(primaryResponse.getResponse().getResult(),
+                equalTo(DocWriteResponse.Result.NOOP));
     }
 
     public void testUpdateReplicaRequestWithFailure() throws Exception {
-        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id").source(Requests.INDEX_CONTENT_TYPE, "field", "value");
+        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id")
+                .source(Requests.INDEX_CONTENT_TYPE, "field", "value");
         BulkItemRequest replicaRequest = new BulkItemRequest(0, writeRequest);
 
         Exception err = new ElasticsearchException("I'm dead <(x.x)>");
         Engine.IndexResult indexResult = new Engine.IndexResult(err, 0, 0);
-        BulkItemResultHolder failedResults = new BulkItemResultHolder(null, indexResult, replicaRequest);
+        BulkItemResultHolder failedResults = new BulkItemResultHolder(null, indexResult,
+                replicaRequest);
 
         Translog.Location location = new Translog.Location(0, 0, 0);
         BulkItemRequest[] items = new BulkItemRequest[0];
-        BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
-        Translog.Location newLocation = TransportShardBulkAction.updateReplicaRequest(failedResults,
-                DocWriteRequest.OpType.UPDATE, location, bulkShardRequest);
+        BulkShardRequest bulkShardRequest =
+                new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
+        BulkItemResponse primaryResponse =
+                TransportShardBulkAction.createPrimaryResponse(
+                        failedResults, DocWriteRequest.OpType.UPDATE, bulkShardRequest);
 
-        BulkItemResponse primaryResponse = replicaRequest.getPrimaryResponse();
+        Translog.Location newLocation =
+                TransportShardBulkAction.calculateTranslogLocation(location, failedResults);
 
         // Since this was not a conflict failure, the primary response
         // should be filled out with the failure information
@@ -378,20 +410,26 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
     }
 
     public void testUpdateReplicaRequestWithConflictFailure() throws Exception {
-        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id").source(Requests.INDEX_CONTENT_TYPE, "field", "value");
+        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id")
+                .source(Requests.INDEX_CONTENT_TYPE, "field", "value");
         BulkItemRequest replicaRequest = new BulkItemRequest(0, writeRequest);
 
-        Exception err = new VersionConflictEngineException(shardId, "type", "id", "I'm conflicted <(;_;)>");
+        Exception err = new VersionConflictEngineException(shardId, "type", "id",
+                "I'm conflicted <(;_;)>");
         Engine.IndexResult indexResult = new Engine.IndexResult(err, 0, 0);
-        BulkItemResultHolder failedResults = new BulkItemResultHolder(null, indexResult, replicaRequest);
+        BulkItemResultHolder failedResults = new BulkItemResultHolder(null, indexResult,
+                replicaRequest);
 
         Translog.Location location = new Translog.Location(0, 0, 0);
         BulkItemRequest[] items = new BulkItemRequest[0];
-        BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
-        Translog.Location newLocation = TransportShardBulkAction.updateReplicaRequest(failedResults,
-                DocWriteRequest.OpType.UPDATE, location, bulkShardRequest);
+        BulkShardRequest bulkShardRequest =
+                new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
+        BulkItemResponse primaryResponse =
+                TransportShardBulkAction.createPrimaryResponse(
+                        failedResults, DocWriteRequest.OpType.UPDATE, bulkShardRequest);
 
-        BulkItemResponse primaryResponse = replicaRequest.getPrimaryResponse();
+        Translog.Location newLocation =
+                TransportShardBulkAction.calculateTranslogLocation(location, failedResults);
 
         // Since this was not a conflict failure, the primary response
         // should be filled out with the failure information
@@ -410,22 +448,27 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
     }
 
     public void testUpdateReplicaRequestWithSuccess() throws Exception {
-        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id").source(Requests.INDEX_CONTENT_TYPE, "field", "value");
+        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id")
+                .source(Requests.INDEX_CONTENT_TYPE, "field", "value");
         BulkItemRequest replicaRequest = new BulkItemRequest(0, writeRequest);
 
         boolean created = randomBoolean();
         Translog.Location resultLocation = new Translog.Location(42, 42, 42);
         Engine.IndexResult indexResult = new FakeResult(1, 1, created, resultLocation);
         DocWriteResponse indexResponse = new IndexResponse(shardId, "index", "id", 1, 1, created);
-        BulkItemResultHolder goodResults = new BulkItemResultHolder(indexResponse, indexResult, replicaRequest);
+        BulkItemResultHolder goodResults =
+                new BulkItemResultHolder(indexResponse, indexResult, replicaRequest);
 
         Translog.Location originalLocation = new Translog.Location(21, 21, 21);
         BulkItemRequest[] items = new BulkItemRequest[0];
-        BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
-        Translog.Location newLocation = TransportShardBulkAction.updateReplicaRequest(goodResults,
-                DocWriteRequest.OpType.INDEX, originalLocation, bulkShardRequest);
+        BulkShardRequest bulkShardRequest =
+                new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
+        BulkItemResponse primaryResponse =
+                TransportShardBulkAction.createPrimaryResponse(
+                        goodResults, DocWriteRequest.OpType.INDEX, bulkShardRequest);
 
-        BulkItemResponse primaryResponse = replicaRequest.getPrimaryResponse();
+        Translog.Location newLocation =
+                TransportShardBulkAction.calculateTranslogLocation(originalLocation, goodResults);
 
         // Check that the translog is successfully advanced
         assertThat(newLocation, equalTo(resultLocation));
@@ -438,6 +481,61 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
         assertThat(response.status(), equalTo(created ? RestStatus.CREATED : RestStatus.OK));
     }
 
+    public void testCalculateTranslogLocation() throws Exception {
+        final Translog.Location original = new Translog.Location(0, 0, 0);
+
+        DocWriteRequest writeRequest = new IndexRequest("index", "type", "id")
+                .source(Requests.INDEX_CONTENT_TYPE, "field", "value");
+        BulkItemRequest replicaRequest = new BulkItemRequest(0, writeRequest);
+        BulkItemResultHolder results = new BulkItemResultHolder(null, null, replicaRequest);
+
+        assertThat(TransportShardBulkAction.calculateTranslogLocation(original, results),
+                equalTo(original));
+
+        boolean created = randomBoolean();
+        DocWriteResponse indexResponse = new IndexResponse(shardId, "index", "id", 1, 1, created);
+        Translog.Location newLocation = new Translog.Location(1, 1, 1);
+        Engine.IndexResult indexResult = new IndexResultWithLocation(randomNonNegativeLong(),
+                randomNonNegativeLong(), created, newLocation);
+        results = new BulkItemResultHolder(indexResponse, indexResult, replicaRequest);
+        assertThat(TransportShardBulkAction.calculateTranslogLocation(original, results),
+                equalTo(newLocation));
+
+    }
+
+    public class IndexResultWithLocation extends Engine.IndexResult {
+        private final Translog.Location location;
+        public IndexResultWithLocation(long version, long seqNo, boolean created,
+                                       Translog.Location newLocation) {
+            super(version, seqNo, created);
+            this.location = newLocation;
+        }
+
+        @Override
+        public Translog.Location getTranslogLocation() {
+            return this.location;
+        }
+    }
+
+    public void testPrepareIndexOpOnReplica() throws Exception {
+        IndexMetaData metaData = indexMetaData();
+        IndexShard shard = newStartedShard(false);
+
+        DocWriteResponse primaryResponse = new IndexResponse(shardId, "index", "id",
+                1, 1, randomBoolean());
+        IndexRequest request = new IndexRequest("index", "type", "id")
+                .source(Requests.INDEX_CONTENT_TYPE, "field", "value");
+
+        Engine.Index op = TransportShardBulkAction.prepareIndexOperationOnReplica(
+                primaryResponse, request, shard);
+
+        assertThat(op.version(), equalTo(primaryResponse.getVersion()));
+        assertThat(op.seqNo(), equalTo(primaryResponse.getSeqNo()));
+        assertThat(op.versionType(), equalTo(VersionType.EXTERNAL));
+
+        closeShards(shard);
+    }
+
     /**
      * Fake IndexResult that has a settable translog location
      */
@@ -445,7 +543,8 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
 
         private final Translog.Location location;
 
-        protected FakeResult(long version, long seqNo, boolean created, Translog.Location location) {
+        protected FakeResult(long version, long seqNo, boolean created,
+                             Translog.Location location) {
             super(version, seqNo, created);
             this.location = location;
         }
@@ -458,23 +557,12 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
 
     /** Doesn't perform any mapping updates */
     public static class NoopMappingUpdatePerformer implements MappingUpdatePerformer {
-        public MappingUpdatePerformer.MappingUpdateResult updateMappingsIfNeeded(IndexShard primary,
-                                                                                 IndexRequest request) throws Exception {
-            Engine.Index operation = TransportShardBulkAction.prepareIndexOperationOnPrimary(request, primary);
-            return new MappingUpdatePerformer.MappingUpdateResult(operation);
-        }
-    }
-
-    /** Always returns the given failure */
-    private class FailingMappingUpdatePerformer implements MappingUpdatePerformer {
-        private final Exception e;
-        FailingMappingUpdatePerformer(Exception e) {
-            this.e = e;
+        public void updateMappingsIfNeeded(Engine.Index operation,
+                                           ShardId shardId,
+                                           String type) throws Exception {
         }
 
-        public MappingUpdatePerformer.MappingUpdateResult updateMappingsIfNeeded(IndexShard primary,
-                                                                                 IndexRequest request) throws Exception {
-            return new MappingUpdatePerformer.MappingUpdateResult(e);
+        public void verifyMappings(Engine.Index operation, ShardId shardId) throws Exception {
         }
     }
 
@@ -485,8 +573,30 @@ public class TransportShardBulkActionTests extends IndexShardTestCase {
             this.e = e;
         }
 
-        public MappingUpdatePerformer.MappingUpdateResult updateMappingsIfNeeded(IndexShard primary,
-                                                                                 IndexRequest request) throws Exception {
+        public void updateMappingsIfNeeded(Engine.Index operation,
+                                           ShardId shardId,
+                                           String type) throws Exception {
+            throw e;
+        }
+
+        public void verifyMappings(Engine.Index operation, ShardId shardId) throws Exception {
+            fail("should not have gotten to this point");
+        }
+    }
+
+    /** Always throw the given exception */
+    private class ThrowingVerifyingMappingUpdatePerformer implements MappingUpdatePerformer {
+        private final Exception e;
+        ThrowingVerifyingMappingUpdatePerformer(Exception e) {
+            this.e = e;
+        }
+
+        public void updateMappingsIfNeeded(Engine.Index operation,
+                                           ShardId shardId,
+                                           String type) throws Exception {
+        }
+
+        public void verifyMappings(Engine.Index operation, ShardId shardId) throws Exception {
             throw e;
         }
     }