[RECOVERY] Only iterate the files that we recovered from the commit

Today we use Directory#listAll() to find all the files we recovered. Yet,
this is not accurate since there might be leftovers etc. It's better to
only iterate over the known files from the segments info that we recovered.

src/main/java/org/elasticsearch/common/lucene/Lucene.java

@@ -19,19 +19,13 @@
 
 package org.elasticsearch.common.lucene;
 
+import com.google.common.collect.Iterables;
 import org.apache.lucene.analysis.core.KeywordAnalyzer;
 import org.apache.lucene.analysis.standard.StandardAnalyzer;
 import org.apache.lucene.codecs.CodecUtil;
 import org.apache.lucene.codecs.DocValuesFormat;
 import org.apache.lucene.codecs.PostingsFormat;
-import org.apache.lucene.index.CorruptIndexException;
-import org.apache.lucene.index.DirectoryReader;
-import org.apache.lucene.index.IndexCommit;
-import org.apache.lucene.index.IndexFormatTooNewException;
-import org.apache.lucene.index.IndexFormatTooOldException;
-import org.apache.lucene.index.IndexWriter;
-import org.apache.lucene.index.LeafReaderContext;
-import org.apache.lucene.index.SegmentInfos;
+import org.apache.lucene.index.*;
 import org.apache.lucene.search.Collector;
 import org.apache.lucene.search.ComplexExplanation;
 import org.apache.lucene.search.ConstantScoreQuery;
@@ -70,6 +64,10 @@
 
 import java.io.IOException;
 import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
 
 import static org.elasticsearch.common.lucene.search.NoopCollector.NOOP_COLLECTOR;
 
@@ -120,6 +118,17 @@ public static SegmentInfos readSegmentInfos(Directory directory) throws IOExcept
         return SegmentInfos.readLatestCommit(directory);
     }
 
+    /**
+     * Returns an iterable that allows to iterate over all files in this segments info
+     */
+    public static Iterable<String> files(SegmentInfos infos) throws IOException {
+        final List<Collection<String>> list = new ArrayList<>();
+        for (SegmentCommitInfo info : infos) {
+            list.add(info.files());
+        }
+        return Iterables.concat(list.toArray(new Collection[0]));
+    }
+
     /**
      * Reads the segments infos from the given commit, failing if it fails to load
      */

src/main/java/org/elasticsearch/index/gateway/IndexShardGateway.java

@@ -23,6 +23,7 @@
 import org.apache.lucene.index.IndexWriter;
 import org.apache.lucene.index.IndexWriterConfig;
 import org.apache.lucene.index.SegmentInfos;
+import org.apache.lucene.store.Directory;
 import org.apache.lucene.util.IOUtils;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
@@ -106,11 +107,11 @@ public void recover(boolean indexShouldExists, RecoveryState recoveryState) thro
         long version = -1;
         long translogId = -1;
         final Set<String> typesToUpdate = Sets.newHashSet();
+        SegmentInfos si = null;
         indexShard.store().incRef();
         try {
             try {
                 indexShard.store().failIfCorrupted();
-                SegmentInfos si = null;
                 try {
                     si = Lucene.readSegmentInfos(indexShard.store().directory());
                 } catch (Throwable e) {
@@ -156,11 +157,14 @@ public void recover(boolean indexShouldExists, RecoveryState recoveryState) thro
             try {
                 int numberOfFiles = 0;
                 long totalSizeInBytes = 0;
-                for (String name : indexShard.store().directory().listAll()) {
-                    numberOfFiles++;
-                    long length =  indexShard.store().directory().fileLength(name);
-                    totalSizeInBytes += length;
-                    recoveryState.getIndex().addFileDetail(name, length, length);
+                if (si != null) {
+                    final Directory directory = indexShard.store().directory();
+                    for (String name : Lucene.files(si)) {
+                        numberOfFiles++;
+                        long length = directory.fileLength(name);
+                        totalSizeInBytes += length;
+                        recoveryState.getIndex().addFileDetail(name, length, length);
+                    }
                 }
                 RecoveryState.Index index = recoveryState.getIndex();
                 index.totalFileCount(numberOfFiles);