Snapshot/Restore: keep the last legacy checksums file at the end of r…

…estore

 This commit fixes the issue caused by restore process deleting all legacy checksum files at the end of restore process. Instead it keeps the latest version of the checksum intact. The issue manifests itself in losing checksum for all legacy files restored into post 1.3.0 cluster, which in turn causes unnecessary snapshotting of files that didn't change.

Fixes #8119

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

@@ -802,7 +802,7 @@ public void restore() throws IOException {
                 /// now, go over and clean files that are in the store, but were not in the snapshot
                 try {
                     for (String storeFile : store.directory().listAll()) {
-                        if (!snapshot.containPhysicalIndexFile(storeFile)) {
+                        if (!Store.isChecksum(storeFile) && !snapshot.containPhysicalIndexFile(storeFile)) {
                             try {
                                 store.directory().deleteFile(storeFile);
                             } catch (IOException e) {

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

@@ -39,6 +39,7 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.lucene.Directories;
 import org.elasticsearch.common.lucene.Lucene;
 import org.elasticsearch.common.lucene.store.InputStreamIndexInput;
@@ -609,6 +610,8 @@ public String toString() {
      * @see StoreFileMetaData
      */
     public final static class MetadataSnapshot implements Iterable<StoreFileMetaData> {
+        private static final ESLogger logger = Loggers.getLogger(MetadataSnapshot.class);
+
         private final Map<String, StoreFileMetaData> metadata;
 
         public static final MetadataSnapshot EMPTY = new MetadataSnapshot();
@@ -672,6 +675,16 @@ ImmutableMap<String, StoreFileMetaData> buildMetadata(IndexCommit commit, Direct
             return builder.build();
         }
 
+        /**
+         * Reads legacy checksum files found in the directory.
+         *
+         * Files are expected to start with _checksums- prefix
+         * followed by long file version. Only file with the highest version is read, all other files are ignored.
+         *
+         * @param directory the directory to read checksums from
+         * @return a map of file checksums and the checksum file version
+         * @throws IOException
+         */
         static Tuple<Map<String, String>, Long> readLegacyChecksums(Directory directory) throws IOException {
             synchronized (directory) {
                 long lastFound = -1;
@@ -694,6 +707,30 @@ static Tuple<Map<String, String>, Long> readLegacyChecksums(Directory directory)
             }
         }
 
+        /**
+         * Deletes all checksum files with version lower than newVersion.
+         *
+         * @param directory the directory to clean
+         * @param newVersion the latest checksum file version
+         * @throws IOException
+         */
+        static void cleanLegacyChecksums(Directory directory, long newVersion) throws IOException {
+            synchronized (directory) {
+                for (String name : directory.listAll()) {
+                    if (isChecksum(name)) {
+                        long current = Long.parseLong(name.substring(CHECKSUMS_PREFIX.length()));
+                        if (current < newVersion) {
+                            try {
+                                directory.deleteFile(name);
+                            } catch (IOException ex) {
+                                logger.debug("can't delete old checksum file [{}]", ex, name);
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
         private static void checksumFromLuceneFile(Directory directory, String file, ImmutableMap.Builder<String, StoreFileMetaData> builder,  ESLogger logger, Version version, boolean readFileAsHash) throws IOException {
             final String checksum;
             final BytesRefBuilder fileHash = new BytesRefBuilder();
@@ -904,6 +941,7 @@ synchronized void writeChecksums(Directory directory, Map<String, String> checks
                 output.writeStringStringMap(checksums);
             }
             directory.sync(Collections.singleton(checksumName));
+            MetadataSnapshot.cleanLegacyChecksums(directory, nextVersion);
         }
 
         public void clear() {