Recovery from local gateway should re-introduce new mappings

The delayed mapping intro tests exposed a bug where if a new mapping is introduced, yet not updated on the master, and a full restart occurs, reply of the transaction log will not cause the new mapping to be re-introduced.
closes elastic#6659

add comment on the method

Author: kimchy

src/main/java/org/elasticsearch/index/gateway/local/LocalIndexShardGateway.java

@@ -24,13 +24,16 @@
 import org.apache.lucene.index.SegmentInfos;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.cluster.action.index.MappingUpdatedAction;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.InputStreamStreamInput;
 import org.elasticsearch.common.lucene.Lucene;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.gateway.IndexShardGateway;
 import org.elasticsearch.index.gateway.IndexShardGatewayRecoveryException;
+import org.elasticsearch.index.service.IndexService;
 import org.elasticsearch.index.settings.IndexSettings;
 import org.elasticsearch.index.shard.AbstractIndexShardComponent;
 import org.elasticsearch.index.shard.IndexShardState;
@@ -57,7 +60,8 @@
 public class LocalIndexShardGateway extends AbstractIndexShardComponent implements IndexShardGateway {
 
     private final ThreadPool threadPool;
-
+    private final MappingUpdatedAction mappingUpdatedAction;
+    private final IndexService indexService;
     private final InternalIndexShard indexShard;
 
     private final RecoveryState recoveryState = new RecoveryState();
@@ -66,9 +70,12 @@ public class LocalIndexShardGateway extends AbstractIndexShardComponent implemen
     private final TimeValue syncInterval;
 
     @Inject
-    public LocalIndexShardGateway(ShardId shardId, @IndexSettings Settings indexSettings, ThreadPool threadPool, IndexShard indexShard) {
+    public LocalIndexShardGateway(ShardId shardId, @IndexSettings Settings indexSettings, ThreadPool threadPool, MappingUpdatedAction mappingUpdatedAction,
+                                  IndexService indexService, IndexShard indexShard) {
         super(shardId, indexSettings);
         this.threadPool = threadPool;
+        this.mappingUpdatedAction = mappingUpdatedAction;
+        this.indexService = indexService;
         this.indexShard = (InternalIndexShard) indexShard;
 
         syncInterval = componentSettings.getAsTime("sync", TimeValue.timeValueSeconds(5));
@@ -224,7 +231,10 @@ public void recover(boolean indexShouldExists, RecoveryState recoveryState) thro
                     break;
                 }
                 try {
-                    indexShard.performRecoveryOperation(operation);
+                    Engine.IndexingOperation potentialIndexOperation = indexShard.performRecoveryOperation(operation);
+                    if (potentialIndexOperation != null) {
+                        mappingUpdatedAction.updateMappingOnMaster(indexService.index().name(), potentialIndexOperation.docMapper(), indexService.indexUUID());
+                    }
                     recoveryState.getTranslog().addTranslogOperations(1);
                 } catch (ElasticsearchException e) {
                     if (e.status() == RestStatus.BAD_REQUEST) {

src/main/java/org/elasticsearch/index/shard/service/InternalIndexShard.java

@@ -736,24 +736,33 @@ public void performRecoveryFinalization(boolean withFlush) throws ElasticsearchE
         engine.enableGcDeletes(true);
     }
 
-    public void performRecoveryOperation(Translog.Operation operation) throws ElasticsearchException {
+    /**
+     * Performs a single recovery operation, and returns the indexing operation (or null if its not an indexing operation)
+     * that can then be used for mapping updates (for example) if needed.
+     */
+    public Engine.IndexingOperation performRecoveryOperation(Translog.Operation operation) throws ElasticsearchException {
         if (state != IndexShardState.RECOVERING) {
             throw new IndexShardNotRecoveringException(shardId, state);
         }
+        Engine.IndexingOperation indexOperation = null;
         try {
             switch (operation.opType()) {
                 case CREATE:
                     Translog.Create create = (Translog.Create) operation;
-                    engine.create(prepareCreate(
+                    Engine.Create engineCreate = prepareCreate(
                             source(create.source()).type(create.type()).id(create.id())
-                            .routing(create.routing()).parent(create.parent()).timestamp(create.timestamp()).ttl(create.ttl()),
-                            create.version(), create.versionType().versionTypeForReplicationAndRecovery(), Engine.Operation.Origin.RECOVERY, true, false));
+                                    .routing(create.routing()).parent(create.parent()).timestamp(create.timestamp()).ttl(create.ttl()),
+                            create.version(), create.versionType().versionTypeForReplicationAndRecovery(), Engine.Operation.Origin.RECOVERY, true, false);
+                    engine.create(engineCreate);
+                    indexOperation = engineCreate;
                     break;
                 case SAVE:
                     Translog.Index index = (Translog.Index) operation;
-                    engine.index(prepareIndex(source(index.source()).type(index.type()).id(index.id())
-                            .routing(index.routing()).parent(index.parent()).timestamp(index.timestamp()).ttl(index.ttl()),
-                            index.version(),index.versionType().versionTypeForReplicationAndRecovery(), Engine.Operation.Origin.RECOVERY, true));
+                    Engine.Index engineIndex = prepareIndex(source(index.source()).type(index.type()).id(index.id())
+                                    .routing(index.routing()).parent(index.parent()).timestamp(index.timestamp()).ttl(index.ttl()),
+                            index.version(), index.versionType().versionTypeForReplicationAndRecovery(), Engine.Operation.Origin.RECOVERY, true);
+                    engine.index(engineIndex);
+                    indexOperation = engineIndex;
                     break;
                 case DELETE:
                     Translog.Delete delete = (Translog.Delete) operation;
@@ -786,6 +795,7 @@ public void performRecoveryOperation(Translog.Operation operation) throws Elasti
                 throw e;
             }
         }
+        return indexOperation;
     }
 
     /**