[ENGINE] Make translog fully private to the engine

This commit moves the translog creation into the InternalEngine
to ensure the transactino log is created after we acquired the write
lock on the index. This also prevents races when ShadowEngines are shutting
down due to node restarts where another node already takes over the not yet
fully synced transaction log.

Author: s1monw

src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -278,7 +278,7 @@ public final Searcher acquireSearcher(String source) throws EngineException {
     }
 
     /** returns the translog for this engine */
-    public abstract Translog translog();
+    public abstract Translog getTranslog();
 
     protected void ensureOpen() {
         if (isClosed.get()) {

src/main/java/org/elasticsearch/index/engine/EngineConfig.java

@@ -29,6 +29,8 @@
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.BigArray;
+import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.codec.CodecService;
 import org.elasticsearch.index.deletionpolicy.SnapshotDeletionPolicy;
@@ -39,9 +41,12 @@
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.shard.TranslogRecoveryPerformer;
 import org.elasticsearch.index.store.Store;
+import org.elasticsearch.index.translog.fs.FsTranslog;
 import org.elasticsearch.indices.IndicesWarmer;
 import org.elasticsearch.threadpool.ThreadPool;
 
+import java.io.IOException;
+import java.nio.file.Path;
 import java.util.concurrent.TimeUnit;
 
 /*
@@ -77,6 +82,8 @@ public final class EngineConfig {
     private final boolean ignoreUnknownTranslog;
     private final QueryCache filterCache;
     private final QueryCachingPolicy filterCachingPolicy;
+    private final BigArrays bigArrays;
+    private final Path translogPath;
 
     /**
      * Index setting for index concurrency / number of threadstates in the indexwriter.
@@ -139,10 +146,10 @@ public final class EngineConfig {
      * Creates a new {@link org.elasticsearch.index.engine.EngineConfig}
      */
     public EngineConfig(ShardId shardId, ThreadPool threadPool, ShardIndexingService indexingService,
-            IndexSettingsService indexSettingsService, IndicesWarmer warmer, Store store, SnapshotDeletionPolicy deletionPolicy,
-            MergePolicyProvider mergePolicyProvider, MergeSchedulerProvider mergeScheduler, Analyzer analyzer,
-            Similarity similarity, CodecService codecService, Engine.FailedEngineListener failedEngineListener,
-            TranslogRecoveryPerformer translogRecoveryPerformer, QueryCache filterCache, QueryCachingPolicy filterCachingPolicy) {
+                        IndexSettingsService indexSettingsService, IndicesWarmer warmer, Store store, SnapshotDeletionPolicy deletionPolicy,
+                        MergePolicyProvider mergePolicyProvider, MergeSchedulerProvider mergeScheduler, Analyzer analyzer,
+                        Similarity similarity, CodecService codecService, Engine.FailedEngineListener failedEngineListener,
+                        TranslogRecoveryPerformer translogRecoveryPerformer, QueryCache filterCache, QueryCachingPolicy filterCachingPolicy, BigArrays bigArrays, Path translogPath) {
         this.shardId = shardId;
         this.threadPool = threadPool;
         this.indexingService = indexingService;
@@ -156,6 +163,8 @@ public EngineConfig(ShardId shardId, ThreadPool threadPool, ShardIndexingService
         this.similarity = similarity;
         this.codecService = codecService;
         this.failedEngineListener = failedEngineListener;
+        this.bigArrays = bigArrays;
+        this.translogPath = translogPath;
         Settings indexSettings = indexSettingsService.getSettings();
         this.optimizeAutoGenerateId = indexSettings.getAsBoolean(EngineConfig.INDEX_OPTIMIZE_AUTOGENERATED_ID_SETTING, false);
         this.compoundOnFlush = indexSettings.getAsBoolean(EngineConfig.INDEX_COMPOUND_ON_FLUSH, compoundOnFlush);
@@ -421,4 +430,25 @@ public QueryCache getFilterCache() {
     public QueryCachingPolicy getFilterCachingPolicy() {
         return filterCachingPolicy;
     }
+
+    /**
+     * Returns a BigArrays instance for this engine
+     */
+    public BigArrays getBigArrays() {
+        return bigArrays;
+    }
+
+    /**
+     * Returns the translog path for this engine
+     */
+    public Path getTranslogPath() {
+        return translogPath;
+    }
+
+    /**
+     * Returns the {@link org.elasticsearch.index.settings.IndexSettingsService} for this engine.
+     */
+    public IndexSettingsService getIndesSettingService() {
+        return indexSettingsService;
+    }
 }

src/main/java/org/elasticsearch/index/engine/EngineFactory.java

@@ -25,7 +25,7 @@
  */
 public interface EngineFactory {
 
-    public Engine newReadWriteEngine(EngineConfig config, FsTranslog translog, boolean skipTranslogRecovery);
+    public Engine newReadWriteEngine(EngineConfig config, boolean skipTranslogRecovery);
 
     public Engine newReadOnlyEngine(EngineConfig config);
 }

src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -104,19 +104,18 @@ public class InternalEngine extends Engine {
 
     private final IndexThrottle throttle;
 
-    public InternalEngine(EngineConfig engineConfig, FsTranslog translog, boolean skipInitialTranslogRecovery) throws EngineException {
+    public InternalEngine(EngineConfig engineConfig, boolean skipInitialTranslogRecovery) throws EngineException {
         super(engineConfig);
-        Preconditions.checkNotNull(translog, "Translog must be provided to the engine");
         this.versionMap = new LiveVersionMap();
         store.incRef();
         IndexWriter writer = null;
+        FsTranslog translog = null;
         SearcherManager manager = null;
         boolean success = false;
         try {
             this.lastDeleteVersionPruneTimeMSec = engineConfig.getThreadPool().estimatedTimeInMillis();
             this.indexingService = engineConfig.getIndexingService();
             this.warmer = engineConfig.getWarmer();
-            this.translog = translog;
             this.mergePolicyProvider = engineConfig.getMergePolicyProvider();
             this.mergeScheduler = engineConfig.getMergeScheduler();
             this.dirtyLocks = new Object[engineConfig.getIndexConcurrency() * 50]; // we multiply it to have enough...
@@ -130,10 +129,12 @@ public InternalEngine(EngineConfig engineConfig, FsTranslog translog, boolean sk
             try {
                 writer = createWriter();
                 indexWriter = writer;
+                translog = new FsTranslog(engineConfig.getShardId(), engineConfig.getIndesSettingService(), engineConfig.getBigArrays(), engineConfig.getTranslogPath(), engineConfig.getThreadPool());
                 committedTranslogId = loadCommittedTranslogId(writer, translog);
             } catch (IOException e) {
                 throw new EngineCreationFailureException(shardId, "failed to create engine", e);
             }
+            this.translog = translog;
             manager = createSearcherManager();
             this.searcherManager = manager;
             this.versionMap.setManager(searcherManager);
@@ -154,7 +155,7 @@ public InternalEngine(EngineConfig engineConfig, FsTranslog translog, boolean sk
             success = true;
         } finally {
             if (success == false) {
-                IOUtils.closeWhileHandlingException(writer, manager);
+                IOUtils.closeWhileHandlingException(writer, translog, manager);
                 versionMap.clear();
                 if (isClosed.get() == false) {
                     // failure we need to dec the store reference
@@ -166,7 +167,7 @@ public InternalEngine(EngineConfig engineConfig, FsTranslog translog, boolean sk
     }
 
     @Override
-    public Translog translog() {
+    public Translog getTranslog() {
         ensureOpen();
         return translog;
     }
@@ -913,7 +914,6 @@ protected final void closeNoLock(String reason) {
             assert rwl.isWriteLockedByCurrentThread() || failEngineLock.isHeldByCurrentThread() : "Either the write lock must be held or the engine must be currently be failing itself";
             try {
                 this.versionMap.clear();
-                logger.trace("close searcherManager");
                 try {
                     IOUtils.close(searcherManager);
                 } catch (Throwable t) {

src/main/java/org/elasticsearch/index/engine/InternalEngineFactory.java

@@ -22,8 +22,8 @@
 
 public class InternalEngineFactory implements EngineFactory {
     @Override
-    public Engine newReadWriteEngine(EngineConfig config, FsTranslog translog, boolean skipTranslogRecovery) {
-        return new InternalEngine(config, translog, skipTranslogRecovery);
+    public Engine newReadWriteEngine(EngineConfig config, boolean skipTranslogRecovery) {
+        return new InternalEngine(config, skipTranslogRecovery);
     }
 
     @Override

src/main/java/org/elasticsearch/index/engine/ShadowEngine.java

@@ -169,8 +169,8 @@ public GetResult get(Get get) throws EngineException {
     }
 
     @Override
-    public Translog translog() {
-        throw new UnsupportedOperationException("shard engines don't have translogs");
+    public Translog getTranslog() {
+        throw new UnsupportedOperationException("shadow engines don't have translogs");
     }
 
     @Override

src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -101,7 +101,6 @@
 import org.elasticsearch.index.termvectors.ShardTermVectorsService;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.index.translog.TranslogStats;
-import org.elasticsearch.index.translog.fs.FsTranslog;
 import org.elasticsearch.index.warmer.ShardIndexWarmerService;
 import org.elasticsearch.index.warmer.WarmerStats;
 import org.elasticsearch.indices.IndicesLifecycle;
@@ -259,8 +258,8 @@ public boolean canIndex() {
         return true;
     }
 
-    public Translog translog() {
-        return engine().translog();
+    public Translog.View newTranslogView() {
+        return engine().getTranslog().newView();
     }
 
     public ShardIndexingService indexingService() {
@@ -655,7 +654,7 @@ public IdCacheStats idCacheStats() {
     }
 
     public TranslogStats translogStats() {
-        return engine().translog().stats();
+        return engine().getTranslog().stats();
     }
 
     public SuggestStats suggestStats() {
@@ -817,12 +816,18 @@ private Map<String, Mapping> internalPerformTranslogRecovery(boolean skipTranslo
      * After the store has been recovered, we need to start the engine. This method starts a new engine but skips
      * the replay of the transaction log which is required in cases where we restore a previous index or recover from
      * a remote peer.
+     *
+     * @param wipeTranslogs if set to <code>true</code> all skipped / uncommitted translogs are removed.
      */
-    public void skipTranslogRecovery() {
+    public void skipTranslogRecovery(boolean wipeTranslogs) throws IOException {
         assert engineUnsafe() == null : "engine was already created";
         Map<String, Mapping> recoveredTypes = internalPerformTranslogRecovery(true);
         assert recoveredTypes.isEmpty();
         assert recoveryState.getTranslog().recoveredOperations() == 0;
+        if (wipeTranslogs) {
+            final Translog translog = engine().getTranslog();
+            translog.markCommitted(translog.currentId());
+        }
     }
 
     /** called if recovery has to be restarted after network error / delay ** */
@@ -964,7 +969,7 @@ public void updateBufferSize(ByteSizeValue shardIndexingBufferSize, ByteSizeValu
         }
         Engine engine = engineUnsafe();
         if (engine != null) {
-            engine.translog().updateBuffer(shardTranslogBufferSize);
+            engine.getTranslog().updateBuffer(shardTranslogBufferSize);
         }
     }
 
@@ -1218,21 +1223,7 @@ private void createNewEngine(boolean skipTranslogRecovery, EngineConfig config)
     }
 
     protected Engine newEngine(boolean skipTranslogRecovery, EngineConfig config) {
-        final FsTranslog translog;
-        try {
-            translog = new FsTranslog(shardId, indexSettingsService, bigArrays, path, threadPool);
-        } catch (IOException e) {
-            throw new EngineCreationFailureException(shardId, "failed to create translog", e);
-        }
-        Engine engine = null;
-        try {
-            engine = engineFactory.newReadWriteEngine(config, translog, skipTranslogRecovery);
-        } finally {
-            if (engine == null) {
-                IOUtils.closeWhileHandlingException(translog);
-            }
-        }
-        return engine;
+        return engineFactory.newReadWriteEngine(config, skipTranslogRecovery);
     }
 
     /**
@@ -1293,6 +1284,6 @@ protected void operationProcessed() {
         };
         return new EngineConfig(shardId,
                 threadPool, indexingService, indexSettingsService, warmer, store, deletionPolicy, mergePolicyProvider, mergeScheduler,
-                mapperAnalyzer, similarityService.similarity(), codecService, failedEngineListener, translogRecoveryPerformer, indexCache.filter(), indexCache.filterPolicy());
+                mapperAnalyzer, similarityService.similarity(), codecService, failedEngineListener, translogRecoveryPerformer, indexCache.filter(), indexCache.filterPolicy(), bigArrays, shardPath().resolveTranslog());
     }
 }

src/main/java/org/elasticsearch/index/shard/ShadowIndexShard.java

@@ -121,10 +121,4 @@ protected Engine newEngine(boolean skipInitialTranslogRecovery, EngineConfig con
     public boolean allowsPrimaryPromotion() {
         return false;
     }
-
-    @Override
-    @Nullable
-    public Translog translog() {
-        throw new UnsupportedOperationException("shadow shards don't have a translog");
-    }
 }

src/main/java/org/elasticsearch/index/snapshots/IndexShardSnapshotAndRestoreService.java

@@ -125,8 +125,7 @@ public void restore(final RecoveryState recoveryState) {
                 snapshotShardId = new ShardId(restoreSource.index(), shardId.id());
             }
             indexShardRepository.restore(restoreSource.snapshotId(), shardId, snapshotShardId, recoveryState);
-            indexShard.skipTranslogRecovery();
-            indexShard.translog().markCommitted(indexShard.translog().currentId());
+            indexShard.skipTranslogRecovery(true);
             indexShard.finalizeRecovery();
             indexShard.postRecovery("restore done");
             restoreService.indexShardRestoreCompleted(restoreSource.snapshotId(), shardId);

src/main/java/org/elasticsearch/index/translog/TranslogService.java

@@ -142,7 +142,7 @@ public void run() {
                 return;
             }
 
-            if (indexShard.translog() == null) {
+            if (indexShard.engine().getTranslog() == null) {
                 reschedule();
                 return;
             }