Introduce index.version.compatibility setting

server/src/main/java/org/elasticsearch/cluster/coordination/JoinTaskExecutor.java

@@ -272,22 +272,22 @@ public static void ensureIndexCompatibility(final Version nodeVersion, Metadata
         // we ensure that all indices in the cluster we join are compatible with us no matter if they are
         // closed or not we can't read mappings of these indices so we need to reject the join...
         for (IndexMetadata idxMetadata : metadata) {
-            if (idxMetadata.getCreationVersion().after(nodeVersion)) {
+            if (idxMetadata.getCompatibilityVersion().after(nodeVersion)) {
                 throw new IllegalStateException(
                     "index "
                         + idxMetadata.getIndex()
                         + " version not supported: "
-                        + idxMetadata.getCreationVersion()
+                        + idxMetadata.getCompatibilityVersion()
                         + " the node version is: "
                         + nodeVersion
                 );
             }
-            if (idxMetadata.getCreationVersion().before(supportedIndexVersion)) {
+            if (idxMetadata.getCompatibilityVersion().before(supportedIndexVersion)) {
                 throw new IllegalStateException(
                     "index "
                         + idxMetadata.getIndex()
                         + " version not supported: "
-                        + idxMetadata.getCreationVersion()
+                        + idxMetadata.getCompatibilityVersion()
                         + " minimum compatible index version is: "
                         + supportedIndexVersion
                 );

server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java

@@ -336,6 +336,49 @@ public static APIBlock readFrom(StreamInput input) throws IOException {
     @Deprecated
     public static final String SETTING_VERSION_UPGRADED_STRING = "index.version.upgraded_string";
 
+    public static final String SETTING_VERSION_COMPATIBILITY = "index.version.compatibility";
+
+    /**
+     * See {@link #getCompatibilityVersion()}
+     */
+    public static final Setting<Version> SETTING_INDEX_VERSION_COMPATIBILITY = Setting.versionSetting(
+        SETTING_VERSION_COMPATIBILITY,
+        SETTING_INDEX_VERSION_CREATED, // fall back to index.version.created
+        new Setting.Validator<>() {
+
+            @Override
+            public void validate(final Version compatibilityVersion) {
+
+            }
+
+            @Override
+            public void validate(final Version compatibilityVersion, final Map<Setting<?>, Object> settings) {
+                Version createdVersion = (Version) settings.get(SETTING_INDEX_VERSION_CREATED);
+                if (compatibilityVersion.before(createdVersion)) {
+                    throw new IllegalArgumentException(
+                        SETTING_VERSION_COMPATIBILITY
+                            + " ["
+                            + compatibilityVersion
+                            + "] must be >= "
+                            + SETTING_VERSION_CREATED
+                            + " ["
+                            + createdVersion
+                            + "]"
+                    );
+                }
+            }
+
+            @Override
+            public Iterator<Setting<?>> settings() {
+                final List<Setting<?>> settings = List.of(SETTING_INDEX_VERSION_CREATED);
+                return settings.iterator();
+            }
+
+        },
+        Property.IndexScope,
+        Property.PrivateIndex
+    );
+
     /**
      * The user provided name for an index. This is the plain string provided by the user when the index was created.
      * It might still contain date math expressions etc. (added in 5.0)
@@ -484,6 +527,7 @@ public static APIBlock readFrom(StreamInput input) throws IOException {
     private final DiscoveryNodeFilters initialRecoveryFilters;
 
     private final Version indexCreatedVersion;
+    private final Version indexCompatibilityVersion;
 
     private final ActiveShardCount waitForActiveShards;
     private final ImmutableOpenMap<String, RolloverInfo> rolloverInfos;
@@ -592,6 +636,7 @@ private IndexMetadata(
         this.autoExpandReplicas = autoExpandReplicas;
         this.isSearchableSnapshot = isSearchableSnapshot;
         this.isPartialSearchableSnapshot = isPartialSearchableSnapshot;
+        this.indexCompatibilityVersion = SETTING_INDEX_VERSION_COMPATIBILITY.get(settings);
         assert numberOfShards * routingFactor == routingNumShards : routingNumShards + " must be a multiple of " + numberOfShards;
     }
 
@@ -689,11 +734,23 @@ public long primaryTerm(int shardId) {
     /**
      * Return the {@link Version} on which this index has been created. This
      * information is typically useful for backward compatibility.
+     * To check index compatibility (e.g. N-1 checks), use {@link #getCompatibilityVersion()} instead.
      */
     public Version getCreationVersion() {
         return indexCreatedVersion;
     }
 
+    /**
+     * Return the {@link Version} that this index provides compatibility for.
+     * This is typically compared to the {@link Version#minimumIndexCompatibilityVersion()} to figure out whether the index can be handled
+     * by the cluster.
+     * By default, this is equal to the {@link #getCreationVersion()}, but can also be a newer version if the index has been imported as
+     * a legacy index from an older snapshot, and its metadata has been converted to be handled by newer version nodes.
+     */
+    public Version getCompatibilityVersion() {
+        return indexCompatibilityVersion;
+    }
+
     public long getCreationDate() {
         return creationDate;
     }
@@ -1923,11 +1980,12 @@ public static IndexMetadata legacyFromXContent(XContentParser parser) throws IOE
                 } else if (token == XContentParser.Token.START_OBJECT) {
                     if ("settings".equals(currentFieldName)) {
                         Settings settings = Settings.fromXContent(parser);
-                        if (SETTING_INDEX_VERSION_CREATED.get(settings).onOrAfter(Version.CURRENT.minimumIndexCompatibilityVersion())) {
+                        if (SETTING_INDEX_VERSION_COMPATIBILITY.get(settings)
+                            .onOrAfter(Version.CURRENT.minimumIndexCompatibilityVersion())) {
                             throw new IllegalStateException(
-                                "this method should only be used to parse older index metadata versions "
+                                "this method should only be used to parse older incompatible index metadata versions "
                                     + "but got "
-                                    + SETTING_INDEX_VERSION_CREATED.get(settings)
+                                    + SETTING_INDEX_VERSION_COMPATIBILITY.get(settings)
                             );
                         }
                         builder.settings(settings);
@@ -2008,7 +2066,7 @@ public static IndexMetadata legacyFromXContent(XContentParser parser) throws IOE
             }
 
             IndexMetadata indexMetadata = builder.build();
-            assert indexMetadata.getCreationVersion().before(Version.CURRENT.minimumIndexCompatibilityVersion());
+            assert indexMetadata.getCompatibilityVersion().before(Version.CURRENT.minimumIndexCompatibilityVersion());
             return indexMetadata;
         }
 

server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadataVerifier.java

@@ -99,13 +99,13 @@ public IndexMetadata verifyIndexMetadata(IndexMetadata indexMetadata, Version mi
      * previous major version.
      */
     private void checkSupportedVersion(IndexMetadata indexMetadata, Version minimumIndexCompatibilityVersion) {
-        boolean isSupportedVersion = indexMetadata.getCreationVersion().onOrAfter(minimumIndexCompatibilityVersion);
+        boolean isSupportedVersion = indexMetadata.getCompatibilityVersion().onOrAfter(minimumIndexCompatibilityVersion);
         if (isSupportedVersion == false) {
             throw new IllegalStateException(
                 "The index "
                     + indexMetadata.getIndex()
-                    + " was created with version ["
-                    + indexMetadata.getCreationVersion()
+                    + " has current compatibility version ["
+                    + indexMetadata.getCompatibilityVersion()
                     + "] but the minimum compatible version is ["
                     + minimumIndexCompatibilityVersion
                     + "]. It should be re-indexed in Elasticsearch "

server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java

@@ -1663,7 +1663,7 @@ public Metadata build(boolean builtIndicesLookupEagerly) {
                     }
                 }
                 indexMetadata.getAliases().keysIt().forEachRemaining(indicesAliases::add);
-                oldestIndexVersionId = Math.min(oldestIndexVersionId, indexMetadata.getCreationVersion().id);
+                oldestIndexVersionId = Math.min(oldestIndexVersionId, indexMetadata.getCompatibilityVersion().id);
             }
 
             final DataStreamMetadata dataStreamMetadata = (DataStreamMetadata) this.customs.get(DataStreamMetadata.TYPE);

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java

@@ -1463,6 +1463,7 @@ static void prepareResizeIndexSettings(
         }
 
         indexSettingsBuilder.put(IndexMetadata.SETTING_VERSION_CREATED, sourceMetadata.getCreationVersion())
+            .put(IndexMetadata.SETTING_VERSION_COMPATIBILITY, sourceMetadata.getCompatibilityVersion())
             .put(builder.build())
             .put(IndexMetadata.SETTING_ROUTING_PARTITION_SIZE, sourceMetadata.getRoutingPartitionSize())
             .put(IndexMetadata.INDEX_RESIZE_SOURCE_NAME.getKey(), resizeSourceIndex.getName())

server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -53,6 +53,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         MergeSchedulerConfig.MAX_MERGE_COUNT_SETTING,
         MergeSchedulerConfig.MAX_THREAD_COUNT_SETTING,
         IndexMetadata.SETTING_INDEX_VERSION_CREATED,
+        IndexMetadata.SETTING_INDEX_VERSION_COMPATIBILITY,
         IndexMetadata.INDEX_ROUTING_EXCLUDE_GROUP_SETTING,
         IndexMetadata.INDEX_ROUTING_INCLUDE_GROUP_SETTING,
         IndexMetadata.INDEX_ROUTING_REQUIRE_GROUP_SETTING,

server/src/main/java/org/elasticsearch/common/settings/Setting.java

@@ -1233,6 +1233,15 @@ public static Setting<Version> versionSetting(final String key, final Version de
         return new Setting<>(key, s -> Integer.toString(defaultValue.id), s -> Version.fromId(Integer.parseInt(s)), properties);
     }
 
+    public static Setting<Version> versionSetting(
+        final String key,
+        Setting<Version> fallbackSetting,
+        Validator<Version> validator,
+        Property... properties
+    ) {
+        return new Setting<>(key, fallbackSetting, s -> Version.fromId(Integer.parseInt(s)), properties);
+    }
+
     public static Setting<Float> floatSetting(String key, float defaultValue, Property... properties) {
         return new Setting<>(key, (s) -> Float.toString(defaultValue), Float::parseFloat, properties);
     }

server/src/main/java/org/elasticsearch/gateway/LocalAllocateDangledIndices.java

@@ -120,24 +120,24 @@ public ClusterState execute(ClusterState currentState) {
                     boolean importNeeded = false;
                     StringBuilder sb = new StringBuilder();
                     for (IndexMetadata indexMetadata : request.indices) {
-                        if (indexMetadata.getCreationVersion().before(minIndexCompatibilityVersion)) {
+                        if (indexMetadata.getCompatibilityVersion().before(minIndexCompatibilityVersion)) {
                             logger.warn(
-                                "ignoring dangled index [{}] on node [{}] since it's created version [{}] is not supported by at "
-                                    + "least one node in the cluster minVersion [{}]",
+                                "ignoring dangled index [{}] on node [{}] since it's current compatibility version [{}] "
+                                    + "is not supported by at least one node in the cluster minVersion [{}]",
                                 indexMetadata.getIndex(),
                                 request.fromNode,
-                                indexMetadata.getCreationVersion(),
+                                indexMetadata.getCompatibilityVersion(),
                                 minIndexCompatibilityVersion
                             );
                             continue;
                         }
-                        if (currentState.nodes().getMinNodeVersion().before(indexMetadata.getCreationVersion())) {
+                        if (currentState.nodes().getMinNodeVersion().before(indexMetadata.getCompatibilityVersion())) {
                             logger.warn(
-                                "ignoring dangled index [{}] on node [{}]"
-                                    + " since its created version [{}] is later than the oldest versioned node in the cluster [{}]",
+                                "ignoring dangled index [{}] on node [{}] since its current compatibility version [{}] "
+                                    + "is later than the oldest versioned node in the cluster [{}]",
                                 indexMetadata.getIndex(),
                                 request.fromNode,
-                                indexMetadata.getCreationVersion(),
+                                indexMetadata.getCompatibilityVersion(),
                                 currentState.getNodes().getMasterNode().getVersion()
                             );
                             continue;

server/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -938,6 +938,16 @@ public synchronized boolean updateIndexMetadata(IndexMetadata indexMetadata) {
         if (version.equals(newIndexVersion) == false) {
             throw new IllegalArgumentException("version mismatch on settings update expected: " + version + " but was: " + newIndexVersion);
         }
+        Version newCompatibilityVersion = IndexMetadata.SETTING_INDEX_VERSION_COMPATIBILITY.get(newSettings);
+        Version compatibilityVersion = IndexMetadata.SETTING_INDEX_VERSION_COMPATIBILITY.get(settings);
+        if (compatibilityVersion.equals(newCompatibilityVersion) == false) {
+            throw new IllegalArgumentException(
+                "compatibility version mismatch on settings update expected: "
+                    + compatibilityVersion
+                    + " but was: "
+                    + newCompatibilityVersion
+            );
+        }
         final String newUUID = newSettings.get(IndexMetadata.SETTING_INDEX_UUID, IndexMetadata.INDEX_UUID_NA_VALUE);
         if (newUUID.equals(getUUID()) == false) {
             throw new IllegalArgumentException("uuid mismatch on settings update expected: " + getUUID() + " but was: " + newUUID);

server/src/main/java/org/elasticsearch/snapshots/RestoreService.java

@@ -1285,9 +1285,7 @@ public ClusterState execute(ClusterState currentState) {
 
             final ImmutableOpenMap.Builder<ShardId, ShardRestoreStatus> shardsBuilder = ImmutableOpenMap.builder();
 
-            final Version minIndexCompatibilityVersion = skipVersionChecks(repositoryMetadata)
-                ? Version.fromString("1.0.0")
-                : currentState.getNodes().getMaxNodeVersion().minimumIndexCompatibilityVersion();
+            final Version minIndexCompatibilityVersion = currentState.getNodes().getMaxNodeVersion().minimumIndexCompatibilityVersion();
             final String localNodeId = clusterService.state().nodes().getLocalNodeId();
             for (Map.Entry<String, IndexId> indexEntry : indicesToRestore.entrySet()) {
                 final IndexId index = indexEntry.getValue();
@@ -1297,10 +1295,9 @@ public ClusterState execute(ClusterState currentState) {
                     request.indexSettings(),
                     request.ignoreIndexSettings()
                 );
-                if (snapshotIndexMetadata.getCreationVersion()
-                    .before(currentState.getNodes().getMaxNodeVersion().minimumIndexCompatibilityVersion())) {
+                if (snapshotIndexMetadata.getCompatibilityVersion().before(minIndexCompatibilityVersion)) {
                     // adapt index metadata so that it can be understood by current version
-                    snapshotIndexMetadata = convertLegacyIndex(snapshotIndexMetadata);
+                    snapshotIndexMetadata = convertLegacyIndex(snapshotIndexMetadata, currentState);
                 }
                 try {
                     snapshotIndexMetadata = indexMetadataVerifier.verifyIndexMetadata(snapshotIndexMetadata, minIndexCompatibilityVersion);
@@ -1590,7 +1587,10 @@ public void clusterStateProcessed(ClusterState oldState, ClusterState newState)
         }
     }
 
-    private IndexMetadata convertLegacyIndex(IndexMetadata snapshotIndexMetadata) {
+    private IndexMetadata convertLegacyIndex(IndexMetadata snapshotIndexMetadata, ClusterState clusterState) {
+        if (snapshotIndexMetadata.getCreationVersion().before(Version.fromString("5.0.0"))) {
+            throw new IllegalArgumentException("can't restore an index created before version 5.0.0");
+        }
         MappingMetadata mappingMetadata = snapshotIndexMetadata.mapping();
         Map<String, Object> loadedMappingSource = mappingMetadata.rawSourceAsMap();
 
@@ -1613,7 +1613,17 @@ private IndexMetadata convertLegacyIndex(IndexMetadata snapshotIndexMetadata) {
         Map<String, Object> newMapping = new LinkedHashMap<>();
         newMapping.put(mappingMetadata.type(), newMappingSource);
         // TODO: _routing? Perhaps we don't need to obey any routing here as stuff is read-only anyway and get API will be disabled
-        return IndexMetadata.builder(snapshotIndexMetadata).putMapping(new MappingMetadata(mappingMetadata.type(), newMapping)).build();
+        return IndexMetadata.builder(snapshotIndexMetadata)
+            .putMapping(new MappingMetadata(mappingMetadata.type(), newMapping))
+            .settings(
+                Settings.builder()
+                    .put(snapshotIndexMetadata.getSettings())
+                    .put(
+                        IndexMetadata.SETTING_INDEX_VERSION_COMPATIBILITY.getKey(),
+                        clusterState.getNodes().getSmallestNonClientNodeVersion()
+                    )
+            )
+            .build();
     }
 
     private static IndexMetadata.Builder restoreToCreateNewIndex(IndexMetadata snapshotIndexMetadata, String renamedIndexName) {

server/src/test/java/org/elasticsearch/cluster/coordination/JoinTaskExecutorTests.java

@@ -110,17 +110,13 @@ public void testSuccess() {
         Settings.builder().build();
         Metadata.Builder metaBuilder = Metadata.builder();
         IndexMetadata indexMetadata = IndexMetadata.builder("test")
-            .settings(
-                settings(VersionUtils.randomVersionBetween(random(), Version.CURRENT.minimumIndexCompatibilityVersion(), Version.CURRENT))
-            )
+            .settings(randomCompatibleVersionSettings())
             .numberOfShards(1)
             .numberOfReplicas(1)
             .build();
         metaBuilder.put(indexMetadata, false);
         indexMetadata = IndexMetadata.builder("test1")
-            .settings(
-                settings(VersionUtils.randomVersionBetween(random(), Version.CURRENT.minimumIndexCompatibilityVersion(), Version.CURRENT))
-            )
+            .settings(randomCompatibleVersionSettings())
             .numberOfShards(1)
             .numberOfReplicas(1)
             .build();
@@ -129,6 +125,24 @@ public void testSuccess() {
         JoinTaskExecutor.ensureIndexCompatibility(Version.CURRENT, metadata);
     }
 
+    public static Settings.Builder randomCompatibleVersionSettings() {
+        Settings.Builder builder = Settings.builder();
+        if (randomBoolean()) {
+            builder.put(IndexMetadata.SETTING_VERSION_CREATED, getRandomCompatibleVersion());
+            if (randomBoolean()) {
+                builder.put(IndexMetadata.SETTING_VERSION_COMPATIBILITY, getRandomCompatibleVersion());
+            }
+        } else {
+            builder.put(IndexMetadata.SETTING_VERSION_CREATED, randomFrom(Version.fromString("5.0.0"), Version.fromString("6.0.0")));
+            builder.put(IndexMetadata.SETTING_VERSION_COMPATIBILITY, getRandomCompatibleVersion());
+        }
+        return builder;
+    }
+
+    private static Version getRandomCompatibleVersion() {
+        return VersionUtils.randomVersionBetween(random(), Version.CURRENT.minimumIndexCompatibilityVersion(), Version.CURRENT);
+    }
+
     public void testUpdatesNodeWithNewRoles() throws Exception {
         // Node roles vary by version, and new roles are suppressed for BWC. This means we can receive a join from a node that's already
         // in the cluster but with a different set of roles: the node didn't change roles, but the cluster state came via an older master.