Ensure 5.x indices have 6.x commit tags before open engine

The previous commit chose to fallback the default value if
max_unsafe_autoid_timestamp is not available. That approach is
embarrassing as it may hide an underlying problem. This commit backs out
that logic and ensures all required 6.x commit tags are available (if
not, generate them) before opening engine.

Author: dnhatn

server/src/main/java/org/elasticsearch/index/shard/LocalShardSnapshot.java

@@ -34,7 +34,6 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Collection;
-import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 final class LocalShardSnapshot implements Closeable {
@@ -67,8 +66,7 @@ long maxSeqNo() {
     }
 
     long maxUnsafeAutoIdTimestamp() {
-        final Map<String, String> commitUserData = shard.getEngine().commitStats().getUserData();
-        return Long.parseLong(commitUserData.getOrDefault(InternalEngine.MAX_UNSAFE_AUTO_ID_TIMESTAMP_COMMIT_ID, "-1"));
+        return Long.parseLong(shard.getEngine().commitStats().getUserData().get(InternalEngine.MAX_UNSAFE_AUTO_ID_TIMESTAMP_COMMIT_ID));
     }
 
     Directory getSnapshotDirectory() {

server/src/main/java/org/elasticsearch/index/shard/StoreRecovery.java

@@ -397,7 +397,7 @@ private void internalRecoverFromStore(IndexShard indexShard) throws IndexShardRe
                 store.associateIndexWithNewTranslog(translogUUID);
             } else if (indexShouldExists) {
                 if (indexShard.indexSettings().getIndexVersionCreated().before(Version.V_6_0_0_rc1)) {
-                    if (store.ensureIndexHasHistoryUUIDAndSeqNo()) {
+                    if (store.ensureIndexHas6xCommitTags()) {
                         si = store.readLastCommittedSegmentsInfo(); // new commit is flushed - refresh SegmentInfo.
                     }
                 }

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

@@ -1524,7 +1524,7 @@ public void associateIndexWithNewTranslog(final String translogUUID) throws IOEx
      *
      * @return <code>true</code> if a new commit is flushed, otherwise return false
      */
-    public boolean ensureIndexHasHistoryUUIDAndSeqNo() throws IOException {
+    public boolean ensureIndexHas6xCommitTags() throws IOException {
         metadataLock.writeLock().lock();
         try (IndexWriter writer = newIndexWriter(IndexWriterConfig.OpenMode.APPEND, directory, null)) {
             final Map<String, String> userData = getUserData(writer);
@@ -1538,6 +1538,9 @@ public boolean ensureIndexHasHistoryUUIDAndSeqNo() throws IOException {
                 maps.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(SequenceNumbers.NO_OPS_PERFORMED));
                 maps.put(SequenceNumbers.LOCAL_CHECKPOINT_KEY, Long.toString(SequenceNumbers.NO_OPS_PERFORMED));
             }
+            if (userData.containsKey(InternalEngine.MAX_UNSAFE_AUTO_ID_TIMESTAMP_COMMIT_ID) == false) {
+                userData.put(InternalEngine.MAX_UNSAFE_AUTO_ID_TIMESTAMP_COMMIT_ID, "-1");
+            }
             if (maps.isEmpty() == false) {
                 updateCommitData(writer, maps);
                 return true;

server/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java

@@ -438,7 +438,7 @@ public void cleanFiles(int totalTranslogOps, Store.MetadataSnapshot sourceMetaDa
         try {
             store.cleanupAndVerify("recovery CleanFilesRequestHandler", sourceMetaData);
             if (indexShard.indexSettings().getIndexVersionCreated().before(Version.V_6_0_0_rc1)) {
-                store.ensureIndexHasHistoryUUIDAndSeqNo();
+                store.ensureIndexHas6xCommitTags();
             }
             // TODO: Assign the global checkpoint to the max_seqno of the safe commit if the index version >= 6.2
             final String translogUUID =

server/src/test/java/org/elasticsearch/index/store/StoreTests.java

@@ -1092,7 +1092,7 @@ public void testEnsureIndexHasHistoryUUIDAndSeqNo() throws IOException {
                 writer.commit();
             }
 
-            store.ensureIndexHasHistoryUUIDAndSeqNo();
+            store.ensureIndexHas6xCommitTags();
             SegmentInfos segmentInfos = Lucene.readSegmentInfos(store.directory());
             assertThat(segmentInfos.getUserData(), hasKey(Engine.HISTORY_UUID_KEY));
             assertThat(segmentInfos.getUserData(), hasEntry(equalTo(SequenceNumbers.LOCAL_CHECKPOINT_KEY), equalTo("-1")));
@@ -1110,7 +1110,7 @@ public void testEnsureIndexHasHistoryUUIDAndSeqNo() throws IOException {
                 writer.setLiveCommitData(newCommitData.entrySet());
                 writer.commit();
             }
-            store.ensureIndexHasHistoryUUIDAndSeqNo();
+            store.ensureIndexHas6xCommitTags();
             segmentInfos = Lucene.readSegmentInfos(store.directory());
             assertThat(segmentInfos.getUserData(),
                 hasEntry(equalTo(Engine.HISTORY_UUID_KEY), equalTo(historyUUID)));
@@ -1126,7 +1126,7 @@ public void testEnsureIndexHasHistoryUUIDAndSeqNo() throws IOException {
                 writer.setLiveCommitData(newCommitData.entrySet());
                 writer.commit();
             }
-            store.ensureIndexHasHistoryUUIDAndSeqNo();
+            store.ensureIndexHas6xCommitTags();
             segmentInfos = Lucene.readSegmentInfos(store.directory());
             assertThat(segmentInfos.getUserData(), hasKey(Engine.HISTORY_UUID_KEY));
             assertThat(segmentInfos.getUserData(),