Introduce index settings version (#34429)

This commit introduces settings version to index metadata. This value is
monotonically increasing and is updated on settings updates. This will
be useful in cross-cluster replication so that we can request settings
updates from the leader only when there is a settings update.

server/src/main/java/org/elasticsearch/cluster/ClusterState.java

@@ -284,7 +284,10 @@ public class ClusterState implements ToXContentFragment, Diffable<ClusterState>
         final String TAB = "   ";
         for (IndexMetaData indexMetaData : metaData) {
             sb.append(TAB).append(indexMetaData.getIndex());
-            sb.append(": v[").append(indexMetaData.getVersion()).append("], mv[").append(indexMetaData.getMappingVersion()).append("]\n");
+            sb.append(": v[").append(indexMetaData.getVersion())
+                    .append("], mv[").append(indexMetaData.getMappingVersion())
+                    .append("], sv[").append(indexMetaData.getSettingsVersion())
+                    .append("]\n");
             for (int shard = 0; shard < indexMetaData.getNumberOfShards(); shard++) {
                 sb.append(TAB).append(TAB).append(shard).append(": ");
                 sb.append("p_term [").append(indexMetaData.primaryTerm(shard)).append("], ");

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

@@ -243,6 +243,7 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
     public static final String KEY_IN_SYNC_ALLOCATIONS = "in_sync_allocations";
     static final String KEY_VERSION = "version";
     static final String KEY_MAPPING_VERSION = "mapping_version";
+    static final String KEY_SETTINGS_VERSION = "settings_version";
     static final String KEY_ROUTING_NUM_SHARDS = "routing_num_shards";
     static final String KEY_SETTINGS = "settings";
     static final String KEY_STATE = "state";
@@ -264,6 +265,8 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
 
     private final long mappingVersion;
 
+    private final long settingsVersion;
+
     private final long[] primaryTerms;
 
     private final State state;
@@ -291,7 +294,7 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
     private final ActiveShardCount waitForActiveShards;
     private final ImmutableOpenMap<String, RolloverInfo> rolloverInfos;
 
-    private IndexMetaData(Index index, long version, long mappingVersion, long[] primaryTerms, State state, int numberOfShards, int numberOfReplicas, Settings settings,
+    private IndexMetaData(Index index, long version, long mappingVersion, long settingsVersion, long[] primaryTerms, State state, int numberOfShards, int numberOfReplicas, Settings settings,
                           ImmutableOpenMap<String, MappingMetaData> mappings, ImmutableOpenMap<String, AliasMetaData> aliases,
                           ImmutableOpenMap<String, DiffableStringMap> customData, ImmutableOpenIntMap<Set<String>> inSyncAllocationIds,
                           DiscoveryNodeFilters requireFilters, DiscoveryNodeFilters initialRecoveryFilters, DiscoveryNodeFilters includeFilters, DiscoveryNodeFilters excludeFilters,
@@ -302,6 +305,8 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
         this.version = version;
         assert mappingVersion >= 0 : mappingVersion;
         this.mappingVersion = mappingVersion;
+        assert settingsVersion >= 0 : settingsVersion;
+        this.settingsVersion = settingsVersion;
         this.primaryTerms = primaryTerms;
         assert primaryTerms.length == numberOfShards;
         this.state = state;
@@ -355,6 +360,10 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
         return mappingVersion;
     }
 
+    public long getSettingsVersion() {
+        return settingsVersion;
+    }
+
     /**
      * The term of the current selected primary. This is a non-negative number incremented when
      * a primary shard is assigned after a full cluster restart or a replica shard is promoted to a primary.
@@ -596,6 +605,7 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
         private final int routingNumShards;
         private final long version;
         private final long mappingVersion;
+        private final long settingsVersion;
         private final long[] primaryTerms;
         private final State state;
         private final Settings settings;
@@ -609,6 +619,7 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
             index = after.index.getName();
             version = after.version;
             mappingVersion = after.mappingVersion;
+            settingsVersion = after.settingsVersion;
             routingNumShards = after.routingNumShards;
             state = after.state;
             settings = after.settings;
@@ -630,6 +641,11 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
             } else {
                 mappingVersion = 1;
             }
+            if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+                settingsVersion = in.readVLong();
+            } else {
+                settingsVersion = 1;
+            }
             state = State.fromId(in.readByte());
             settings = Settings.readSettingsFromStream(in);
             primaryTerms = in.readVLongArray();
@@ -658,6 +674,9 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
             if (out.getVersion().onOrAfter(Version.V_6_5_0)) {
                 out.writeVLong(mappingVersion);
             }
+            if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+                out.writeVLong(settingsVersion);
+            }
             out.writeByte(state.id);
             Settings.writeSettingsToStream(settings, out);
             out.writeVLongArray(primaryTerms);
@@ -675,6 +694,7 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
             Builder builder = builder(index);
             builder.version(version);
             builder.mappingVersion(mappingVersion);
+            builder.settingsVersion(settingsVersion);
             builder.setRoutingNumShards(routingNumShards);
             builder.state(state);
             builder.settings(settings);
@@ -696,6 +716,11 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
         } else {
             builder.mappingVersion(1);
         }
+        if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            builder.settingsVersion(in.readVLong());
+        } else {
+            builder.settingsVersion(1);
+        }
         builder.setRoutingNumShards(in.readInt());
         builder.state(State.fromId(in.readByte()));
         builder.settings(readSettingsFromStream(in));
@@ -745,6 +770,9 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
         if (out.getVersion().onOrAfter(Version.V_6_5_0)) {
             out.writeVLong(mappingVersion);
         }
+        if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            out.writeVLong(settingsVersion);
+        }
         out.writeInt(routingNumShards);
         out.writeByte(state.id());
         writeSettingsToStream(settings, out);
@@ -793,6 +821,7 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
         private State state = State.OPEN;
         private long version = 1;
         private long mappingVersion = 1;
+        private long settingsVersion = 1;
         private long[] primaryTerms = null;
         private Settings settings = Settings.Builder.EMPTY_SETTINGS;
         private final ImmutableOpenMap.Builder<String, MappingMetaData> mappings;
@@ -816,6 +845,7 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
             this.state = indexMetaData.state;
             this.version = indexMetaData.version;
             this.mappingVersion = indexMetaData.mappingVersion;
+            this.settingsVersion = indexMetaData.settingsVersion;
             this.settings = indexMetaData.getSettings();
             this.primaryTerms = indexMetaData.primaryTerms.clone();
             this.mappings = ImmutableOpenMap.builder(indexMetaData.mappings);
@@ -990,11 +1020,20 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
             return mappingVersion;
         }
 
+        public long settingsVersion() {
+            return settingsVersion;
+        }
+
         public Builder mappingVersion(final long mappingVersion) {
             this.mappingVersion = mappingVersion;
             return this;
         }
 
+        public Builder settingsVersion(final long settingsVersion) {
+            this.settingsVersion = settingsVersion;
+            return this;
+        }
+
         /**
          * returns the primary term for the given shard.
          * See {@link IndexMetaData#primaryTerm(int)} for more information.
@@ -1122,7 +1161,7 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
 
             final String uuid = settings.get(SETTING_INDEX_UUID, INDEX_UUID_NA_VALUE);
 
-            return new IndexMetaData(new Index(index, uuid), version, mappingVersion, primaryTerms, state, numberOfShards, numberOfReplicas, tmpSettings, mappings.build(),
+            return new IndexMetaData(new Index(index, uuid), version, mappingVersion, settingsVersion, primaryTerms, state, numberOfShards, numberOfReplicas, tmpSettings, mappings.build(),
                 tmpAliases.build(), customMetaData.build(), filledInSyncAllocationIds.build(), requireFilters, initialRecoveryFilters, includeFilters, excludeFilters,
                 indexCreatedVersion, indexUpgradedVersion, getRoutingNumShards(), routingPartitionSize, waitForActiveShards, rolloverInfos.build());
         }
@@ -1132,6 +1171,7 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
 
             builder.field(KEY_VERSION, indexMetaData.getVersion());
             builder.field(KEY_MAPPING_VERSION, indexMetaData.getMappingVersion());
+            builder.field(KEY_SETTINGS_VERSION, indexMetaData.getSettingsVersion());
             builder.field(KEY_ROUTING_NUM_SHARDS, indexMetaData.getRoutingNumShards());
             builder.field(KEY_STATE, indexMetaData.getState().toString().toLowerCase(Locale.ENGLISH));
 
@@ -1205,6 +1245,7 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
                 throw new IllegalArgumentException("expected object but got a " + token);
             }
             boolean mappingVersion = false;
+            boolean settingsVersion = false;
             while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
                 if (token == XContentParser.Token.FIELD_NAME) {
                     currentFieldName = parser.currentName();
@@ -1299,6 +1340,9 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
                     } else if (KEY_MAPPING_VERSION.equals(currentFieldName)) {
                         mappingVersion = true;
                         builder.mappingVersion(parser.longValue());
+                    } else if (KEY_SETTINGS_VERSION.equals(currentFieldName)) {
+                        settingsVersion = true;
+                        builder.settingsVersion(parser.longValue());
                     } else if (KEY_ROUTING_NUM_SHARDS.equals(currentFieldName)) {
                         builder.setRoutingNumShards(parser.intValue());
                     } else {
@@ -1311,6 +1355,9 @@ public class IndexMetaData implements Diffable<IndexMetaData>, ToXContentFragmen
             if (Assertions.ENABLED && Version.indexCreated(builder.settings).onOrAfter(Version.V_6_5_0)) {
                 assert mappingVersion : "mapping version should be present for indices created on or after 6.5.0";
             }
+            if (Assertions.ENABLED && Version.indexCreated(builder.settings).onOrAfter(Version.V_7_0_0_alpha1)) {
+                assert settingsVersion : "settings version should be present for indices created on or after 7.0.0";
+            }
             return builder.build();
         }
     }

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

@@ -51,6 +51,7 @@ import java.util.Map;
 import java.util.Set;
 
 import static org.elasticsearch.action.support.ContextPreservingActionListener.wrapPreservingContext;
+import static org.elasticsearch.index.IndexSettings.same;
 
 /**
  * Service responsible for submitting update index settings requests
@@ -187,6 +188,14 @@ public class MetaDataUpdateSettingsService extends AbstractComponent {
                     }
                 }
 
+                // increment settings versions
+                for (final String index : actualIndices) {
+                    if (same(currentState.metaData().index(index).getSettings(), metaDataBuilder.get(index).getSettings()) == false) {
+                        final IndexMetaData.Builder builder = IndexMetaData.builder(metaDataBuilder.get(index));
+                        builder.settingsVersion(1 + builder.settingsVersion());
+                        metaDataBuilder.put(builder);
+                    }
+                }
 
                 ClusterState updatedState = ClusterState.builder(currentState).metaData(metaDataBuilder).routingTable(routingTableBuilder.build()).blocks(blocks).build();
 
@@ -220,9 +229,9 @@ public class MetaDataUpdateSettingsService extends AbstractComponent {
      */
     private static void maybeUpdateClusterBlock(String[] actualIndices, ClusterBlocks.Builder blocks, ClusterBlock block, Setting<Boolean> setting, Settings openSettings) {
         if (setting.exists(openSettings)) {
-            final boolean updateReadBlock = setting.get(openSettings);
+            final boolean updateBlock = setting.get(openSettings);
             for (String index : actualIndices) {
-                if (updateReadBlock) {
+                if (updateBlock) {
                     blocks.addIndexBlock(index, block);
                 } else {
                     blocks.removeIndexBlock(index, block);

server/src/main/java/org/elasticsearch/cluster/routing/allocation/AllocationService.java

@@ -254,6 +254,13 @@ public class AllocationService extends AbstractComponent {
                 // operation which make these copies stale
                 routingTableBuilder.updateNumberOfReplicas(numberOfReplicas, indices);
                 metaDataBuilder.updateNumberOfReplicas(numberOfReplicas, indices);
+                // update settings version for each index
+                for (final String index : indices) {
+                    final IndexMetaData indexMetaData = metaDataBuilder.get(index);
+                    final IndexMetaData.Builder indexMetaDataBuilder =
+                            new IndexMetaData.Builder(indexMetaData).settingsVersion(1 + indexMetaData.getSettingsVersion());
+                    metaDataBuilder.put(indexMetaDataBuilder);
+                }
                 logger.info("updating number_of_replicas to [{}] for indices {}", numberOfReplicas, indices);
             }
             final ClusterState fixedState = ClusterState.builder(clusterState).routingTable(routingTableBuilder.build())

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

@@ -24,6 +24,8 @@ import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.search.Sort;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.util.Accountable;
+import org.elasticsearch.Assertions;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
@@ -617,9 +619,27 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     }
 
     @Override
-    public synchronized void updateMetaData(final IndexMetaData metadata) {
+    public synchronized void updateMetaData(final IndexMetaData currentIndexMetaData, final IndexMetaData newIndexMetaData) {
         final Translog.Durability oldTranslogDurability = indexSettings.getTranslogDurability();
-        if (indexSettings.updateIndexMetaData(metadata)) {
+
+        final boolean updateIndexMetaData = indexSettings.updateIndexMetaData(newIndexMetaData);
+
+        if (Assertions.ENABLED
+                && currentIndexMetaData != null
+                && currentIndexMetaData.getCreationVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            final long currentSettingsVersion = currentIndexMetaData.getSettingsVersion();
+            final long newSettingsVersion = newIndexMetaData.getSettingsVersion();
+            if (currentSettingsVersion == newSettingsVersion) {
+                assert updateIndexMetaData == false;
+            } else {
+                assert updateIndexMetaData;
+                assert currentSettingsVersion < newSettingsVersion :
+                        "expected current settings version [" + currentSettingsVersion + "] "
+                                + "to be less than new settings version [" + newSettingsVersion + "]";
+            }
+        }
+
+        if (updateIndexMetaData) {
             for (final IndexShard shard : this.shards.values()) {
                 try {
                     shard.onSettingsChanged();

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

@@ -604,17 +604,28 @@ public final class IndexSettings {
             throw new IllegalArgumentException("uuid mismatch on settings update expected: " + getUUID() + " but was: " + newUUID);
         }
         this.indexMetaData = indexMetaData;
-        final Settings existingSettings = this.settings;
+        final Settings newIndexSettings = Settings.builder().put(nodeSettings).put(newSettings).build();
-        if (existingSettings.filter(IndexScopedSettings.INDEX_SETTINGS_KEY_PREDICATE)
+        if (same(this.settings, newIndexSettings)) {
-            .equals(newSettings.filter(IndexScopedSettings.INDEX_SETTINGS_KEY_PREDICATE))) {
             // nothing to update, same settings
             return false;
         }
         scopedSettings.applySettings(newSettings);
-        this.settings = Settings.builder().put(nodeSettings).put(newSettings).build();
+        this.settings = newIndexSettings;
         return true;
     }
 
+    /**
+     * Compare the specified settings for equality.
+     *
+     * @param left  the left settings
+     * @param right the right settings
+     * @return true if the settings are the same, otherwise false
+     */
+    public static boolean same(final Settings left, final Settings right) {
+        return left.filter(IndexScopedSettings.INDEX_SETTINGS_KEY_PREDICATE)
+                .equals(right.filter(IndexScopedSettings.INDEX_SETTINGS_KEY_PREDICATE));
+    }
+
     /**
      * Returns the translog durability for this index.
      */

server/src/main/java/org/elasticsearch/indices/IndicesService.java

@@ -559,7 +559,7 @@ public class IndicesService extends AbstractLifecycleComponent
             closeables.add(() -> service.close("metadata verification", false));
             service.mapperService().merge(metaData, MapperService.MergeReason.MAPPING_RECOVERY);
             if (metaData.equals(metaDataUpdate) == false) {
-                service.updateMetaData(metaDataUpdate);
+                service.updateMetaData(metaData, metaDataUpdate);
             }
         } finally {
             IOUtils.close(closeables);

server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java

@@ -488,7 +488,7 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent imple
             final IndexMetaData newIndexMetaData = state.metaData().index(index);
             assert newIndexMetaData != null : "index " + index + " should have been removed by deleteIndices";
             if (ClusterChangedEvent.indexMetaDataChanged(currentIndexMetaData, newIndexMetaData)) {
-                indexService.updateMetaData(newIndexMetaData);
+                indexService.updateMetaData(currentIndexMetaData, newIndexMetaData);
                 try {
                     if (indexService.updateMapping(currentIndexMetaData, newIndexMetaData) && sendRefreshMapping) {
                         nodeMappingRefreshAction.nodeMappingRefresh(state.nodes().getMasterNode(),
@@ -771,9 +771,12 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent imple
         IndexSettings getIndexSettings();
 
         /**
-         * Updates the meta data of this index. Changes become visible through {@link #getIndexSettings()}
+         * Updates the metadata of this index. Changes become visible through {@link #getIndexSettings()}.
+         *
+         * @param currentIndexMetaData the current index metadata
+         * @param newIndexMetaData the new index metadata
          */
-        void updateMetaData(IndexMetaData indexMetaData);
+        void updateMetaData(IndexMetaData currentIndexMetaData, IndexMetaData newIndexMetaData);
 
         /**
          * Checks if index requires refresh from master.

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

@@ -295,6 +295,7 @@ public class RestoreService extends AbstractComponent implements ClusterStateApp
                                 IndexMetaData.Builder indexMdBuilder = IndexMetaData.builder(snapshotIndexMetaData).state(IndexMetaData.State.OPEN);
                                 indexMdBuilder.version(Math.max(snapshotIndexMetaData.getVersion(), currentIndexMetaData.getVersion() + 1));
                                 indexMdBuilder.mappingVersion(Math.max(snapshotIndexMetaData.getMappingVersion(), currentIndexMetaData.getMappingVersion() + 1));
+                                indexMdBuilder.settingsVersion(Math.max(snapshotIndexMetaData.getSettingsVersion(), currentIndexMetaData.getSettingsVersion() + 1));
                                 if (!request.includeAliases()) {
                                     // Remove all snapshot aliases
                                     if (!snapshotIndexMetaData.getAliases().isEmpty()) {

server/src/test/java/org/elasticsearch/gateway/MetaDataStateFormatTests.java

@@ -264,6 +264,7 @@ public class MetaDataStateFormatTests extends ESTestCase {
             assertThat(deserialized, notNullValue());
             assertThat(deserialized.getVersion(), equalTo(original.getVersion()));
             assertThat(deserialized.getMappingVersion(), equalTo(original.getMappingVersion()));
+            assertThat(deserialized.getSettingsVersion(), equalTo(original.getSettingsVersion()));
             assertThat(deserialized.getNumberOfReplicas(), equalTo(original.getNumberOfReplicas()));
             assertThat(deserialized.getNumberOfShards(), equalTo(original.getNumberOfShards()));
         }

server/src/test/java/org/elasticsearch/indices/cluster/AbstractIndicesClusterStateServiceTestCase.java

@@ -279,10 +279,10 @@ public abstract class AbstractIndicesClusterStateServiceTestCase extends ESTestC
         }
 
         @Override
-        public void updateMetaData(IndexMetaData indexMetaData) {
+        public void updateMetaData(final IndexMetaData currentIndexMetaData, final IndexMetaData newIndexMetaData) {
-            indexSettings.updateIndexMetaData(indexMetaData);
+            indexSettings.updateIndexMetaData(newIndexMetaData);
             for (MockIndexShard shard: shards.values()) {
-                shard.updateTerm(indexMetaData.primaryTerm(shard.shardId().id()));
+                shard.updateTerm(newIndexMetaData.primaryTerm(shard.shardId().id()));
             }
         }
 

server/src/test/java/org/elasticsearch/indices/settings/InternalOrPrivateSettingsPlugin.java

@@ -168,6 +168,7 @@ public class InternalOrPrivateSettingsPlugin extends Plugin implements ActionPlu
                                     .put(currentState.metaData().index(request.index).getSettings())
                                     .put(request.key, request.value);
                     imdBuilder.settings(settingsBuilder);
+                    imdBuilder.settingsVersion(1 + imdBuilder.settingsVersion());
                     builder.put(imdBuilder.build(), true);
                     return ClusterState.builder(currentState).metaData(builder).build();
                 }

server/src/test/java/org/elasticsearch/indices/settings/UpdateNumberOfReplicasIT.java

@@ -72,6 +72,8 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
             assertHitCount(countResponse, 10L);
         }
 
+        final long settingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
         logger.info("Increasing the number of replicas from 1 to 2");
         assertAcked(client().admin().indices().prepareUpdateSettings("test").setSettings(Settings.builder().put("index.number_of_replicas", 2)).execute().actionGet());
         logger.info("Running Cluster Health");
@@ -84,9 +86,16 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         //only 2 copies allocated (1 replica) across 2 nodes
         assertThat(clusterHealth.getIndices().get("test").getActiveShards(), equalTo(numShards.numPrimaries * 2));
 
+        final long afterReplicaIncreaseSettingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        assertThat(afterReplicaIncreaseSettingsVersion, equalTo(1 + settingsVersion));
+
         logger.info("starting another node to new replicas will be allocated to it");
         allowNodes("test", 3);
 
+        final long afterStartingAnotherNodeVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+
         logger.info("Running Cluster Health");
         clusterHealth = client().admin().cluster().prepareHealth().setWaitForEvents(Priority.LANGUID).setWaitForGreenStatus().setWaitForNoRelocatingShards(true).setWaitForNodes(">=3").execute().actionGet();
         logger.info("Done Cluster Health, status {}", clusterHealth.getStatus());
@@ -118,6 +127,10 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         for (int i = 0; i < 10; i++) {
             assertHitCount(client().prepareSearch().setQuery(matchAllQuery()).get(), 10);
         }
+
+        final long afterReplicaDecreaseSettingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        assertThat(afterReplicaDecreaseSettingsVersion, equalTo(1 + afterStartingAnotherNodeVersion));
     }
 
     public void testAutoExpandNumberOfReplicas0ToData() throws IOException {
@@ -136,6 +149,9 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         assertThat(clusterHealth.getIndices().get("test").getNumberOfReplicas(), equalTo(1));
         assertThat(clusterHealth.getIndices().get("test").getActiveShards(), equalTo(numShards.numPrimaries * 2));
 
+        final long settingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+
         logger.info("--> add another node, should increase the number of replicas");
         allowNodes("test", 3);
 
@@ -148,6 +164,10 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         assertThat(clusterHealth.getIndices().get("test").getNumberOfReplicas(), equalTo(2));
         assertThat(clusterHealth.getIndices().get("test").getActiveShards(), equalTo(numShards.numPrimaries * 3));
 
+        final long afterAddingOneNodeSettingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        assertThat(afterAddingOneNodeSettingsVersion, equalTo(1 + settingsVersion));
+
         logger.info("--> closing one node");
         internalCluster().ensureAtMostNumDataNodes(2);
         allowNodes("test", 2);
@@ -161,6 +181,10 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         assertThat(clusterHealth.getIndices().get("test").getNumberOfReplicas(), equalTo(1));
         assertThat(clusterHealth.getIndices().get("test").getActiveShards(), equalTo(numShards.numPrimaries * 2));
 
+        final long afterClosingOneNodeSettingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        assertThat(afterClosingOneNodeSettingsVersion, equalTo(1 + afterAddingOneNodeSettingsVersion));
+
         logger.info("--> closing another node");
         internalCluster().ensureAtMostNumDataNodes(1);
         allowNodes("test", 1);
@@ -173,6 +197,10 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         assertThat(clusterHealth.getIndices().get("test").getActivePrimaryShards(), equalTo(numShards.numPrimaries));
         assertThat(clusterHealth.getIndices().get("test").getNumberOfReplicas(), equalTo(0));
         assertThat(clusterHealth.getIndices().get("test").getActiveShards(), equalTo(numShards.numPrimaries));
+
+        final long afterClosingAnotherNodeSettingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        assertThat(afterClosingAnotherNodeSettingsVersion, equalTo(1 + afterClosingOneNodeSettingsVersion));
     }
 
     public void testAutoExpandNumberReplicas1ToData() throws IOException {
@@ -191,6 +219,8 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         assertThat(clusterHealth.getIndices().get("test").getNumberOfReplicas(), equalTo(1));
         assertThat(clusterHealth.getIndices().get("test").getActiveShards(), equalTo(numShards.numPrimaries * 2));
 
+        final long settingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
         logger.info("--> add another node, should increase the number of replicas");
         allowNodes("test", 3);
 
@@ -203,6 +233,10 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         assertThat(clusterHealth.getIndices().get("test").getNumberOfReplicas(), equalTo(2));
         assertThat(clusterHealth.getIndices().get("test").getActiveShards(), equalTo(numShards.numPrimaries * 3));
 
+        final long afterAddingOneNodeSettingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        assertThat(afterAddingOneNodeSettingsVersion, equalTo(1 + settingsVersion));
+
         logger.info("--> closing one node");
         internalCluster().ensureAtMostNumDataNodes(2);
         allowNodes("test", 2);
@@ -216,6 +250,10 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         assertThat(clusterHealth.getIndices().get("test").getNumberOfReplicas(), equalTo(1));
         assertThat(clusterHealth.getIndices().get("test").getActiveShards(), equalTo(numShards.numPrimaries * 2));
 
+        final long afterClosingOneNodeSettingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        assertThat(afterClosingOneNodeSettingsVersion, equalTo(1 + afterAddingOneNodeSettingsVersion));
+
         logger.info("--> closing another node");
         internalCluster().ensureAtMostNumDataNodes(1);
         allowNodes("test", 1);
@@ -249,6 +287,8 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         allowNodes("test", 4);
         allowNodes("test", 5);
 
+        final long settingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
         logger.info("--> update the auto expand replicas to 0-3");
         client().admin().indices().prepareUpdateSettings("test").setSettings(Settings.builder().put("auto_expand_replicas", "0-3")).execute().actionGet();
 
@@ -260,10 +300,20 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
         assertThat(clusterHealth.getIndices().get("test").getActivePrimaryShards(), equalTo(numShards.numPrimaries));
         assertThat(clusterHealth.getIndices().get("test").getNumberOfReplicas(), equalTo(3));
         assertThat(clusterHealth.getIndices().get("test").getActiveShards(), equalTo(numShards.numPrimaries * 4));
+
+        /*
+         * The settings version increases twice, the first time from the settings update increasing auto expand replicas, and the second
+         * time from the number of replicas changed by the allocation service.
+         */
+        assertThat(
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion(),
+                equalTo(1 + 1 + settingsVersion));
     }
 
     public void testUpdateWithInvalidNumberOfReplicas() {
         createIndex("test");
+        final long settingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
         final int value = randomIntBetween(-10, -1);
         try {
             client().admin().indices().prepareUpdateSettings("test")
@@ -274,6 +324,9 @@ public class UpdateNumberOfReplicasIT extends ESIntegTestCase {
             fail("should have thrown an exception about the replica shard count");
         } catch (IllegalArgumentException e) {
             assertEquals("Failed to parse value [" + value + "] for setting [index.number_of_replicas] must be >= 0", e.getMessage());
+            assertThat(
+                    client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion(),
+                    equalTo(settingsVersion));
         }
     }
 

server/src/test/java/org/elasticsearch/indices/settings/UpdateSettingsIT.java

@@ -475,4 +475,154 @@ public class UpdateSettingsIT extends ESIntegTestCase {
         }
     }
 
+    public void testSettingsVersion() {
+        createIndex("test");
+        ensureGreen("test");
+
+        {
+            final long settingsVersion =
+                    client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+            assertAcked(client()
+                    .admin()
+                    .indices()
+                    .prepareUpdateSettings("test")
+                    .setSettings(Settings.builder().put("index.refresh_interval", "500ms"))
+                    .get());
+            final long newSettingsVersion =
+                    client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+            assertThat(newSettingsVersion, equalTo(1 + settingsVersion));
+        }
+
+        {
+            final boolean block = randomBoolean();
+            assertAcked(client()
+                    .admin()
+                    .indices()
+                    .prepareUpdateSettings("test")
+                    .setSettings(Settings.builder().put("index.blocks.read_only", block))
+                    .get());
+            final long settingsVersion =
+                    client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+            assertAcked(client()
+                    .admin()
+                    .indices()
+                    .prepareUpdateSettings("test")
+                    .setSettings(Settings.builder().put("index.blocks.read_only", block == false))
+                    .get());
+            final long newSettingsVersion =
+                    client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+            assertThat(newSettingsVersion, equalTo(1 + settingsVersion));
+
+            // if the read-only block is present, remove it
+            if (block == false) {
+                assertAcked(client()
+                        .admin()
+                        .indices()
+                        .prepareUpdateSettings("test")
+                        .setSettings(Settings.builder().put("index.blocks.read_only", false))
+                        .get());
+            }
+        }
+    }
+
+    public void testSettingsVersionUnchanged() {
+        createIndex("test");
+        ensureGreen("test");
+
+        {
+            final long settingsVersion =
+                    client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+            final String refreshInterval =
+                    client().admin().indices().prepareGetSettings("test").get().getSetting("test", "index.refresh_interval");
+            assertAcked(client()
+                    .admin()
+                    .indices()
+                    .prepareUpdateSettings("test")
+                    .setSettings(Settings.builder().put("index.refresh_interval", refreshInterval))
+                    .get());
+            final long newSettingsVersion =
+                    client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+            assertThat(newSettingsVersion, equalTo(settingsVersion));
+        }
+
+        {
+            final boolean block = randomBoolean();
+            assertAcked(client()
+                    .admin()
+                    .indices()
+                    .prepareUpdateSettings("test")
+                    .setSettings(Settings.builder().put("index.blocks.read_only", block))
+                    .get());
+            // now put the same block again
+            final long settingsVersion =
+                    client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+            assertAcked(client()
+                    .admin()
+                    .indices()
+                    .prepareUpdateSettings("test")
+                    .setSettings(Settings.builder().put("index.blocks.read_only", block))
+                    .get());
+            final long newSettingsVersion =
+                    client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+            assertThat(newSettingsVersion, equalTo(settingsVersion));
+
+            // if the read-only block is present, remove it
+            if (block) {
+                assertAcked(client()
+                        .admin()
+                        .indices()
+                        .prepareUpdateSettings("test")
+                        .setSettings(Settings.builder().put("index.blocks.read_only", false))
+                        .get());
+            }
+        }
+    }
+
+    /**
+     * The setting {@link IndexMetaData#SETTING_NUMBER_OF_REPLICAS} is special due to handling in
+     * {@link IndexMetaData.Builder#numberOfReplicas(int)}. Therefore we have a dedicated test that this setting is handled properly with
+     * respect to settings version when applying a settings change that does not change the number of replicas.
+     */
+    public void testNumberOfReplicasSettingsVersionUnchanged() {
+        createIndex("test");
+
+        final long settingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        final int numberOfReplicas = Integer.valueOf(
+                client().admin().indices().prepareGetSettings("test").get().getSetting("test", "index.number_of_replicas"));
+        assertAcked(client()
+                .admin()
+                .indices()
+                .prepareUpdateSettings("test")
+                .setSettings(Settings.builder().put("index.number_of_replicas", numberOfReplicas))
+                .get());
+        final long newSettingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        assertThat(newSettingsVersion, equalTo(settingsVersion));
+    }
+
+    /**
+     * The setting {@link IndexMetaData#SETTING_NUMBER_OF_REPLICAS} is special due to handling in
+     * {@link IndexMetaData.Builder#numberOfReplicas(int)}. Therefore we have a dedicated test that this setting is handled properly with
+     * respect to settings version when changing the number of replicas.
+     */
+    public void testNumberOfReplicasSettingsVersion() {
+        createIndex("test");
+
+        final long settingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        final int numberOfReplicas =
+                Integer.valueOf(
+                        client().admin().indices().prepareGetSettings("test").get().getSetting("test", "index.number_of_replicas"));
+        assertAcked(client()
+                .admin()
+                .indices()
+                .prepareUpdateSettings("test")
+                .setSettings(Settings.builder().put("index.number_of_replicas", 1 + numberOfReplicas))
+                .get());
+        final long newSettingsVersion =
+                client().admin().cluster().prepareState().get().getState().metaData().index("test").getSettingsVersion();
+        assertThat(newSettingsVersion, equalTo(1 + settingsVersion));
+    }
+
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/snapshots/SourceOnlySnapshotRepository.java

@@ -94,6 +94,7 @@ public final class SourceOnlySnapshotRepository extends FilterRepository {
                 indexMetadataBuilder.settings(Settings.builder().put(index.getSettings())
                     .put(SOURCE_ONLY.getKey(), true)
                     .put("index.blocks.write", true)); // read-only!
+                indexMetadataBuilder.settingsVersion(1 + indexMetadataBuilder.settingsVersion());
                 builder.put(indexMetadataBuilder);
             }
             super.initializeSnapshot(snapshotId, indices, builder.build());

x-pack/plugin/upgrade/src/main/java/org/elasticsearch/xpack/upgrade/InternalIndexReindexer.java

@@ -27,6 +27,8 @@ import org.elasticsearch.transport.TransportResponse;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
+import static org.elasticsearch.index.IndexSettings.same;
+
 /**
  * A component that performs the following upgrade procedure:
  * <p>
@@ -143,11 +145,17 @@ public class InternalIndexReindexer<T> {
                     throw new IllegalStateException("unable to upgrade a read-only index[" + index + "]");
                 }
 
-                Settings.Builder indexSettings = Settings.builder().put(indexMetaData.getSettings())
+                final Settings indexSettingsBuilder =
-                        .put(IndexMetaData.INDEX_READ_ONLY_SETTING.getKey(), true);
+                        Settings.builder()
+                                .put(indexMetaData.getSettings())
+                                .put(IndexMetaData.INDEX_READ_ONLY_SETTING.getKey(), true)
+                                .build();
+                final IndexMetaData.Builder builder = IndexMetaData.builder(indexMetaData).settings(indexSettingsBuilder);
+                if (same(indexMetaData.getSettings(), indexSettingsBuilder) == false) {
+                    builder.settingsVersion(1 + builder.settingsVersion());
+                }
 
-                MetaData.Builder metaDataBuilder = MetaData.builder(currentState.metaData())
+                MetaData.Builder metaDataBuilder = MetaData.builder(currentState.metaData()).put(builder);
-                        .put(IndexMetaData.builder(indexMetaData).settings(indexSettings));
 
                 ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks())
                         .addIndexBlock(index, IndexMetaData.INDEX_READ_ONLY_BLOCK);