[STORE] Allow to get metadata from arbitrary commit points

Today we always use the latest commit point to return the metadata from
the store. This might cause problems for snapshot and restore since in
contrast to recovery it won't prevent concurrent flushes (lucene commits).
This can lead to all kinds of interesting effects if we are snapshotting
while flushing. This change uses the IndexCommit to open the metadata snapshot
from the store which is consistent with what we snapshot.

Closes #7376

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

@@ -102,6 +102,15 @@ public class Lucene {
         return sis;
     }
 
+    /**
+     * Reads the segments infos from the given commit, failing if it fails to load
+     */
+    public static SegmentInfos readSegmentInfos(IndexCommit commit, Directory directory) throws IOException {
+        final SegmentInfos sis = new SegmentInfos();
+        sis.read(directory, commit.getSegmentsFileName());
+        return sis;
+    }
+
     public static void checkSegmentInfoIntegrity(final Directory directory) throws IOException {
         new SegmentInfos.FindSegmentsFile(directory) {
 

src/main/java/org/elasticsearch/index/snapshots/blobstore/BlobStoreIndexShardRepository.java

@@ -425,7 +425,7 @@ public class BlobStoreIndexShardRepository extends AbstractComponent implements
                 ArrayList<FileInfo> filesToSnapshot = newArrayList();
                 final Store.MetadataSnapshot metadata;
                 try {
-                    metadata = store.getMetadata();
+                    metadata = store.getMetadata(snapshotIndexCommit);
                 } catch (IOException e) {
                     throw new IndexShardSnapshotFailedException(shardId, "Failed to get store file metadata", e);
                 }

src/main/java/org/elasticsearch/index/store/Store.java

@@ -114,16 +114,16 @@ public class Store extends AbstractIndexShardComponent implements CloseableIndex
      * @throws IOException if the index is corrupted or the segments file is not present
      */
     public SegmentInfos readLastCommittedSegmentsInfo() throws IOException {
-        return readLastCommittedSegmentsInfo(directory());
+        return readSegmentsInfo(null, directory());
     }
 
     /**
-     * Returns the last committed segments info for the given directory
+     * Returns the segments info for the given commit or for the latest commit if the given commit is <code>null</code>
      * @throws IOException if the index is corrupted or the segments file is not present
      */
-    private static SegmentInfos readLastCommittedSegmentsInfo(Directory directory) throws IOException {
+    private static SegmentInfos readSegmentsInfo(IndexCommit commit, Directory directory) throws IOException {
         try {
-            return Lucene.readSegmentInfos(directory);
+            return commit == null ? Lucene.readSegmentInfos(directory) : Lucene.readSegmentInfos(commit, directory);
         } catch (EOFException eof) {
             // TODO this should be caught by lucene - EOF is almost certainly an index corruption
             throw new CorruptIndexException("Read past EOF while reading segment infos", eof);
@@ -137,13 +137,21 @@ public class Store extends AbstractIndexShardComponent implements CloseableIndex
     }
 
     /**
-     * Returns a new MetadataSnapshot.
+     * Returns a new MetadataSnapshot for the latest commit in this store.
      */
     public MetadataSnapshot getMetadata() throws IOException {
+        return getMetadata(null);
+    }
+
+    /**
+     * Returns a new MetadataSnapshot for the given commit. If the given commit is <code>null</code>
+     * the latest commit point is used.
+     */
+    public MetadataSnapshot getMetadata(IndexCommit commit) throws IOException {
         ensureOpen();
         failIfCorrupted();
         try {
-            return new MetadataSnapshot(distributorDirectory, logger);
+            return new MetadataSnapshot(commit, distributorDirectory, logger);
         } catch (CorruptIndexException ex) {
             markStoreCorrupted(ex);
             throw ex;
@@ -270,7 +278,7 @@ public class Store extends AbstractIndexShardComponent implements CloseableIndex
             }
             DistributorDirectory dir = new DistributorDirectory(dirs);
             failIfCorrupted(dir, new ShardId("", 1));
-            return new MetadataSnapshot(dir, logger);
+            return new MetadataSnapshot(null, dir, logger);
         } finally {
             IOUtils.close(dirs);
         }
@@ -447,17 +455,17 @@ public class Store extends AbstractIndexShardComponent implements CloseableIndex
     public final static class MetadataSnapshot implements Iterable<StoreFileMetaData> {
         private final ImmutableMap<String, StoreFileMetaData> metadata;
 
-        MetadataSnapshot(Directory directory, ESLogger logger) throws IOException {
-            metadata = buildMetadata(directory, logger);
+        MetadataSnapshot(IndexCommit commit, Directory directory, ESLogger logger) throws IOException {
+            metadata = buildMetadata(commit, directory, logger);
         }
 
-        ImmutableMap<String, StoreFileMetaData> buildMetadata(Directory directory, ESLogger logger) throws IOException {
+        ImmutableMap<String, StoreFileMetaData> buildMetadata(IndexCommit commit, Directory directory, ESLogger logger) throws IOException {
             ImmutableMap.Builder<String, StoreFileMetaData> builder = ImmutableMap.builder();
             Map<String, String> checksumMap = readLegacyChecksums(directory);
             try {
                 final SegmentInfos segmentCommitInfos;
                 try {
-                    segmentCommitInfos = Store.readLastCommittedSegmentsInfo(directory);
+                    segmentCommitInfos = Store.readSegmentsInfo(commit, directory);
                 } catch (FileNotFoundException | NoSuchFileException ex) {
                     // no segments file -- can't read metadata
                     logger.trace("Can't read segment infos", ex);

src/main/java/org/elasticsearch/indices/recovery/RecoverySource.java

@@ -142,7 +142,7 @@ public class RecoverySource extends AbstractComponent {
                 try {
                     StopWatch stopWatch = new StopWatch().start();
                     final Store.MetadataSnapshot metadata;
-                    metadata = store.getMetadata();
+                    metadata = store.getMetadata(snapshot);
                     for (String name : snapshot.getFiles()) {
                         final StoreFileMetaData md = metadata.get(name);
                         if (md == null) {