[Zen2] PersistedState interface implementation (#35819)

Today GatewayMetaState is capable of atomically storing MetaData to
disk. We've also moved fields that are needed to be persisted in Zen2
from ClusterState to ClusterState.MetaData.CoordinationMetaData.

This commit implements PersistedState interface.

version and currentTerm are persisted as a part of Manifest.
GatewayMetaState now implements both ClusterStateApplier and
PersistedState interfaces. We started with two descendants
Zen1GatewayMetaState and Zen2GatewayMetaState, but it turned
out to be not easy to glue it.
GatewayMetaState now constructs previousClusterState (including
MetaData) and previousManifest inside the constructor so that all
PersistedState methods are usable as soon as GatewayMetaState
instance is constructed. Also, loadMetaData is renamed to
getMetaData, because it just returns
previousClusterState.metaData().
Sadly, we don't have access to localNode (obtained from 
TransportService in the constructor, so getLastAcceptedState
should be called, after setLocalNode method is invoked.
Currently, when deciding whether to write IndexMetaData to disk,
we're comparing current IndexMetaData version and received
IndexMetaData version. This is not safe in Zen2 if the term has changed.
So updateClusterState now accepts incremental write
method parameter. When it's set to false, we always write
IndexMetaData to disk.
Things that are not covered by GatewayMetaStateTests are covered
by GatewayMetaStatePersistedStateTests.
This commit also adds an option to use GatewayMetaState instead of
InMemoryPersistedState in TestZenDiscovery. However, by default
InMemoryPersistedState is used and only one test in PersistedStateIT
used GatewayMetaState. In order to use it for other tests, proper
state recovery should be implemented.

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

@@ -458,11 +458,13 @@ public class CoordinationState {
          */
         default void markLastAcceptedConfigAsCommitted() {
             final ClusterState lastAcceptedState = getLastAcceptedState();
-            final CoordinationMetaData coordinationMetaData = CoordinationMetaData.builder(lastAcceptedState.coordinationMetaData())
+            if (lastAcceptedState.getLastAcceptedConfiguration().equals(lastAcceptedState.getLastCommittedConfiguration()) == false) {
-                    .lastCommittedConfiguration(lastAcceptedState.getLastAcceptedConfiguration())
+                final CoordinationMetaData coordinationMetaData = CoordinationMetaData.builder(lastAcceptedState.coordinationMetaData())
-                    .build();
+                        .lastCommittedConfiguration(lastAcceptedState.getLastAcceptedConfiguration())
-            final MetaData metaData = MetaData.builder(lastAcceptedState.metaData()).coordinationMetaData(coordinationMetaData).build();
+                        .build();
-            setLastAcceptedState(ClusterState.builder(lastAcceptedState).metaData(metaData).build());
+                final MetaData metaData = MetaData.builder(lastAcceptedState.metaData()).coordinationMetaData(coordinationMetaData).build();
+                setLastAcceptedState(ClusterState.builder(lastAcceptedState).metaData(metaData).build());
+            }
         }
     }
 

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

@@ -44,16 +44,29 @@ import java.util.stream.Collectors;
  * Index metadata generation could be obtained by calling {@link #getIndexGenerations()}.
  */
 public class Manifest implements ToXContentFragment {
-    private static final long MISSING_GLOBAL_GENERATION = -1;
+    //TODO revisit missing and unknown constants once Zen2 BWC is ready
+    private static final long MISSING_GLOBAL_GENERATION = -1L;
+    private static final long MISSING_CURRENT_TERM = 0L;
+    private static final long UNKNOWN_CURRENT_TERM = MISSING_CURRENT_TERM;
+    private static final long MISSING_CLUSTER_STATE_VERSION = 0L;
+    private static final long UNKNOWN_CLUSTER_STATE_VERSION = MISSING_CLUSTER_STATE_VERSION;
 
     private final long globalGeneration;
     private final Map<Index, Long> indexGenerations;
+    private final long currentTerm;
+    private final long clusterStateVersion;
 
-    public Manifest(long globalGeneration, Map<Index, Long> indexGenerations) {
+    public Manifest(long currentTerm, long clusterStateVersion, long globalGeneration, Map<Index, Long> indexGenerations) {
+        this.currentTerm = currentTerm;
+        this.clusterStateVersion = clusterStateVersion;
         this.globalGeneration = globalGeneration;
         this.indexGenerations = indexGenerations;
     }
 
+    public static Manifest unknownCurrentTermAndVersion(long globalGeneration, Map<Index, Long> indexGenerations) {
+        return new Manifest(UNKNOWN_CURRENT_TERM, UNKNOWN_CLUSTER_STATE_VERSION, globalGeneration, indexGenerations);
+    }
+
     /**
      * Returns global metadata generation.
      */
@@ -68,18 +81,38 @@ public class Manifest implements ToXContentFragment {
         return indexGenerations;
     }
 
+    public long getCurrentTerm() {
+        return currentTerm;
+    }
+
+    public long getClusterStateVersion() {
+        return clusterStateVersion;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         Manifest manifest = (Manifest) o;
-        return globalGeneration == manifest.globalGeneration &&
+        return currentTerm == manifest.currentTerm &&
-                Objects.equals(indexGenerations, manifest.indexGenerations);
+               clusterStateVersion == manifest.clusterStateVersion &&
+               globalGeneration == manifest.globalGeneration &&
+               Objects.equals(indexGenerations, manifest.indexGenerations);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(globalGeneration, indexGenerations);
+        return Objects.hash(currentTerm, clusterStateVersion, globalGeneration, indexGenerations);
+    }
+
+    @Override
+    public String toString() {
+        return "Manifest{" +
+                "currentTerm=" + currentTerm +
+                ", clusterStateVersion=" + clusterStateVersion +
+                ", globalGeneration=" + globalGeneration +
+                ", indexGenerations=" + indexGenerations +
+                '}';
     }
 
     private static final String MANIFEST_FILE_PREFIX = "manifest-";
@@ -103,37 +136,57 @@ public class Manifest implements ToXContentFragment {
      * Code below this comment is for XContent parsing/generation
      */
 
+    private static final ParseField CURRENT_TERM_PARSE_FIELD = new ParseField("current_term");
+    private static final ParseField CLUSTER_STATE_VERSION_PARSE_FIELD = new ParseField("cluster_state_version");
     private static final ParseField GENERATION_PARSE_FIELD = new ParseField("generation");
     private static final ParseField INDEX_GENERATIONS_PARSE_FIELD = new ParseField("index_generations");
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.field(CURRENT_TERM_PARSE_FIELD.getPreferredName(), currentTerm);
+        builder.field(CLUSTER_STATE_VERSION_PARSE_FIELD.getPreferredName(), clusterStateVersion);
         builder.field(GENERATION_PARSE_FIELD.getPreferredName(), globalGeneration);
         builder.array(INDEX_GENERATIONS_PARSE_FIELD.getPreferredName(), indexEntryList().toArray());
         return builder;
     }
 
+    private static long requireNonNullElseDefault(Long value, long defaultValue) {
+        return value != null ? value : defaultValue;
+    }
+
     private List<IndexEntry> indexEntryList() {
         return indexGenerations.entrySet().stream().
                 map(entry -> new IndexEntry(entry.getKey(), entry.getValue())).
                 collect(Collectors.toList());
     }
 
-    private static long generation(Object[] generationAndListOfIndexEntries) {
+    private static long currentTerm(Object[] manifestFields) {
-        return (Long) generationAndListOfIndexEntries[0];
+        return requireNonNullElseDefault((Long) manifestFields[0], MISSING_CURRENT_TERM);
     }
 
-    private static Map<Index, Long> indices(Object[] generationAndListOfIndexEntries) {
+    private static long clusterStateVersion(Object[] manifestFields) {
-        List<IndexEntry> listOfIndices = (List<IndexEntry>) generationAndListOfIndexEntries[1];
+        return requireNonNullElseDefault((Long) manifestFields[1], MISSING_CLUSTER_STATE_VERSION);
+    }
+
+    private static long generation(Object[] manifestFields) {
+        return requireNonNullElseDefault((Long) manifestFields[2], MISSING_GLOBAL_GENERATION);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static Map<Index, Long> indices(Object[] manifestFields) {
+        List<IndexEntry> listOfIndices = (List<IndexEntry>) manifestFields[3];
         return listOfIndices.stream().collect(Collectors.toMap(IndexEntry::getIndex, IndexEntry::getGeneration));
     }
 
     private static final ConstructingObjectParser<Manifest, Void> PARSER = new ConstructingObjectParser<>(
             "manifest",
-            generationAndListOfIndexEntries ->
+            manifestFields ->
-                    new Manifest(generation(generationAndListOfIndexEntries), indices(generationAndListOfIndexEntries)));
+                    new Manifest(currentTerm(manifestFields), clusterStateVersion(manifestFields), generation(manifestFields),
+                            indices(manifestFields)));
 
     static {
+        PARSER.declareLong(ConstructingObjectParser.constructorArg(), CURRENT_TERM_PARSE_FIELD);
+        PARSER.declareLong(ConstructingObjectParser.constructorArg(), CLUSTER_STATE_VERSION_PARSE_FIELD);
         PARSER.declareLong(ConstructingObjectParser.constructorArg(), GENERATION_PARSE_FIELD);
         PARSER.declareObjectArray(ConstructingObjectParser.constructorArg(), IndexEntry.INDEX_ENTRY_PARSER, INDEX_GENERATIONS_PARSE_FIELD);
     }
@@ -143,11 +196,12 @@ public class Manifest implements ToXContentFragment {
     }
 
     public boolean isEmpty() {
-        return globalGeneration == MISSING_GLOBAL_GENERATION && indexGenerations.isEmpty();
+        return currentTerm == MISSING_CURRENT_TERM && clusterStateVersion == MISSING_CLUSTER_STATE_VERSION
+                && globalGeneration == MISSING_GLOBAL_GENERATION && indexGenerations.isEmpty();
     }
 
     public static Manifest empty() {
-        return new Manifest(MISSING_GLOBAL_GENERATION, Collections.emptyMap());
+        return new Manifest(MISSING_CURRENT_TERM, MISSING_CLUSTER_STATE_VERSION, MISSING_GLOBAL_GENERATION, Collections.emptyMap());
     }
 
     public boolean isGlobalGenerationMissing() {

server/src/main/java/org/elasticsearch/discovery/DiscoveryModule.java

@@ -38,6 +38,7 @@ import org.elasticsearch.discovery.zen.FileBasedUnicastHostsProvider;
 import org.elasticsearch.discovery.zen.SettingsBasedHostsProvider;
 import org.elasticsearch.discovery.zen.UnicastHostsProvider;
 import org.elasticsearch.discovery.zen.ZenDiscovery;
+import org.elasticsearch.gateway.GatewayMetaState;
 import org.elasticsearch.plugins.DiscoveryPlugin;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -73,7 +74,7 @@ public class DiscoveryModule {
     public DiscoveryModule(Settings settings, ThreadPool threadPool, TransportService transportService,
                            NamedWriteableRegistry namedWriteableRegistry, NetworkService networkService, MasterService masterService,
                            ClusterApplier clusterApplier, ClusterSettings clusterSettings, List<DiscoveryPlugin> plugins,
-                           AllocationService allocationService, Path configFile) {
+                           AllocationService allocationService, Path configFile, GatewayMetaState gatewayMetaState) {
         final Collection<BiConsumer<DiscoveryNode,ClusterState>> joinValidators = new ArrayList<>();
         final Map<String, Supplier<UnicastHostsProvider>> hostProviders = new HashMap<>();
         hostProviders.put("settings", () -> new SettingsBasedHostsProvider(settings, transportService));
@@ -118,15 +119,16 @@ public class DiscoveryModule {
         Map<String, Supplier<Discovery>> discoveryTypes = new HashMap<>();
         discoveryTypes.put("zen",
             () -> new ZenDiscovery(settings, threadPool, transportService, namedWriteableRegistry, masterService, clusterApplier,
-                clusterSettings, hostsProvider, allocationService, Collections.unmodifiableCollection(joinValidators)));
+                clusterSettings, hostsProvider, allocationService, Collections.unmodifiableCollection(joinValidators), gatewayMetaState));
         discoveryTypes.put("single-node", () -> new SingleNodeDiscovery(settings, transportService, masterService, clusterApplier));
         for (DiscoveryPlugin plugin : plugins) {
             plugin.getDiscoveryTypes(threadPool, transportService, namedWriteableRegistry,
-                masterService, clusterApplier, clusterSettings, hostsProvider, allocationService).entrySet().forEach(entry -> {
+                masterService, clusterApplier, clusterSettings, hostsProvider, allocationService, gatewayMetaState).entrySet()
-                if (discoveryTypes.put(entry.getKey(), entry.getValue()) != null) {
+                    .forEach(entry -> {
-                    throw new IllegalArgumentException("Cannot register discovery type [" + entry.getKey() + "] twice");
+                        if (discoveryTypes.put(entry.getKey(), entry.getValue()) != null) {
-                }
+                            throw new IllegalArgumentException("Cannot register discovery type [" + entry.getKey() + "] twice");
-            });
+                        }
+                    });
         }
         String discoveryType = DISCOVERY_TYPE_SETTING.get(settings);
         Supplier<Discovery> discoverySupplier = discoveryTypes.get(discoveryType);

server/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java

@@ -38,6 +38,7 @@ import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.ClusterApplier;
 import org.elasticsearch.cluster.service.ClusterApplier.ClusterApplyListener;
+import org.elasticsearch.cluster.service.ClusterApplierService;
 import org.elasticsearch.cluster.service.MasterService;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
@@ -58,6 +59,7 @@ import org.elasticsearch.discovery.DiscoverySettings;
 import org.elasticsearch.discovery.DiscoveryStats;
 import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
 import org.elasticsearch.discovery.zen.PublishClusterStateAction.IncomingClusterStateListener;
+import org.elasticsearch.gateway.GatewayMetaState;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.EmptyTransportResponseHandler;
@@ -159,7 +161,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
     public ZenDiscovery(Settings settings, ThreadPool threadPool, TransportService transportService,
                         NamedWriteableRegistry namedWriteableRegistry, MasterService masterService, ClusterApplier clusterApplier,
                         ClusterSettings clusterSettings, UnicastHostsProvider hostsProvider, AllocationService allocationService,
-                        Collection<BiConsumer<DiscoveryNode, ClusterState>> onJoinValidators) {
+                        Collection<BiConsumer<DiscoveryNode, ClusterState>> onJoinValidators, GatewayMetaState gatewayMetaState) {
         super(settings);
         this.onJoinValidators = addBuiltInJoinValidators(onJoinValidators);
         this.masterService = masterService;
@@ -227,6 +229,10 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
 
         transportService.registerRequestHandler(
             DISCOVERY_REJOIN_ACTION_NAME, RejoinClusterRequest::new, ThreadPool.Names.SAME, new RejoinClusterRequestHandler());
+
+        if (clusterApplier instanceof ClusterApplierService) {
+            ((ClusterApplierService) clusterApplier).addLowPriorityApplier(gatewayMetaState);
+        }
     }
 
     static Collection<BiConsumer<DiscoveryNode,ClusterState>> addBuiltInJoinValidators(

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

@@ -24,16 +24,18 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.ClusterChangedEvent;
+import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateApplier;
+import org.elasticsearch.cluster.coordination.CoordinationState;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.Manifest;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.metadata.MetaDataIndexUpgradeService;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.RoutingNode;
 import org.elasticsearch.cluster.routing.ShardRouting;
-import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
@@ -63,23 +65,21 @@ import java.util.function.UnaryOperator;
  * state upgrade if necessary. Also it checks that atomic move is supported on the filesystem level, because it's a must for metadata
  * store algorithm.
  * Please note that the state being loaded when constructing the instance of this class is NOT the state that will be used as a
- * {@link ClusterState#metaData()}. Instead when node is starting up, it calls {@link #loadMetaData()} method and if this node is
+ * {@link ClusterState#metaData()}. Instead when node is starting up, it calls {@link #getMetaData()} method and if this node is
  * elected as master, it requests metaData from other master eligible nodes. After that, master node performs re-conciliation on the
  * gathered results, re-creates {@link ClusterState} and broadcasts this state to other nodes in the cluster.
- * It means that the first time {@link #applyClusterState(ClusterChangedEvent)} method is called, it won't have any previous metaData in
- * memory and will iterate over all the indices in received {@link ClusterState} and store them to disk.
  */
-public class GatewayMetaState implements ClusterStateApplier {
+public class GatewayMetaState implements ClusterStateApplier, CoordinationState.PersistedState {
-    private static final Logger logger = LogManager.getLogger(GatewayMetaState.class);
+    protected static final Logger logger = LogManager.getLogger(GatewayMetaState.class);
 
     private final NodeEnvironment nodeEnv;
-    private final MetaStateService metaStateService;
+    protected final MetaStateService metaStateService;
     private final Settings settings;
 
-    @Nullable
+    //there is a single thread executing updateClusterState calls, hence no volatile modifier
-    //there is a single thread executing applyClusterState calls, hence no volatile modifier
+    protected Manifest previousManifest;
-    private Manifest previousManifest;
+    protected ClusterState previousClusterState;
-    private MetaData previousMetaData;
+    protected boolean incrementalWrite;
 
     public GatewayMetaState(Settings settings, NodeEnvironment nodeEnv, MetaStateService metaStateService,
                             MetaDataIndexUpgradeService metaDataIndexUpgradeService, MetaDataUpgrader metaDataUpgrader) throws IOException {
@@ -90,18 +90,30 @@ public class GatewayMetaState implements ClusterStateApplier {
         ensureNoPre019State(); //TODO remove this check, it's Elasticsearch version 7 already
         ensureAtomicMoveSupported(); //TODO move this check to NodeEnvironment, because it's related to all types of metadata
         upgradeMetaData(metaDataIndexUpgradeService, metaDataUpgrader);
-        profileLoadMetaData();
+        initializeClusterState(ClusterName.CLUSTER_NAME_SETTING.get(settings));
+        incrementalWrite = false;
     }
 
-    private void profileLoadMetaData() throws IOException {
+    private void initializeClusterState(ClusterName clusterName) throws IOException {
-        if (isMasterOrDataNode()) {
+        long startNS = System.nanoTime();
-            long startNS = System.nanoTime();
+        Tuple<Manifest, MetaData> manifestAndMetaData = metaStateService.loadFullState();
-            metaStateService.loadFullState();
+        previousManifest = manifestAndMetaData.v1();
-            logger.debug("took {} to load state", TimeValue.timeValueMillis(TimeValue.nsecToMSec(System.nanoTime() - startNS)));
+
-        }
+        previousClusterState = ClusterState.builder(clusterName)
+                .version(previousManifest.getClusterStateVersion())
+                .metaData(manifestAndMetaData.v2()).build();
+
+        logger.debug("took {} to load state", TimeValue.timeValueMillis(TimeValue.nsecToMSec(System.nanoTime() - startNS)));
     }
 
-    private void upgradeMetaData(MetaDataIndexUpgradeService metaDataIndexUpgradeService, MetaDataUpgrader metaDataUpgrader)
+    public void setLocalNode(DiscoveryNode localNode) {
+        assert previousClusterState.nodes().getLocalNode() == null : "setLocalNode must only be called once";
+        previousClusterState = ClusterState.builder(previousClusterState)
+                .nodes(DiscoveryNodes.builder().add(localNode).localNodeId(localNode.getId()).build())
+                .build();
+    }
+
+    protected void upgradeMetaData(MetaDataIndexUpgradeService metaDataIndexUpgradeService, MetaDataUpgrader metaDataUpgrader)
             throws IOException {
         if (isMasterOrDataNode()) {
             try {
@@ -134,7 +146,8 @@ public class GatewayMetaState implements ClusterStateApplier {
                     }
                 }
 
-                final Manifest newManifest = new Manifest(globalStateGeneration, indices);
+                final Manifest newManifest = new Manifest(manifest.getCurrentTerm(), manifest.getClusterStateVersion(),
+                        globalStateGeneration, indices);
                 writer.writeManifestAndCleanup("startup", newManifest);
             } catch (Exception e) {
                 logger.error("failed to read or upgrade local state, exiting...", e);
@@ -143,7 +156,7 @@ public class GatewayMetaState implements ClusterStateApplier {
         }
     }
 
-    private boolean isMasterOrDataNode() {
+    protected boolean isMasterOrDataNode() {
         return DiscoveryNode.isMasterNode(settings) || DiscoveryNode.isDataNode(settings);
     }
 
@@ -153,8 +166,8 @@ public class GatewayMetaState implements ClusterStateApplier {
         }
     }
 
-    public MetaData loadMetaData() throws IOException {
+    public MetaData getMetaData() {
-        return metaStateService.loadFullState().v2();
+        return previousClusterState.metaData();
     }
 
     @Override
@@ -164,22 +177,55 @@ public class GatewayMetaState implements ClusterStateApplier {
         }
 
         if (event.state().blocks().disableStatePersistence()) {
-            // reset the current state, we need to start fresh...
+            incrementalWrite = false;
-            previousMetaData = null;
-            previousManifest = null;
             return;
         }
 
         try {
-            if (previousManifest == null) {
+            updateClusterState(event.state(), event.previousState());
-                previousManifest = metaStateService.loadManifestOrEmpty();
+            incrementalWrite = true;
-            }
+        } catch (WriteStateException e) {
-            updateMetaData(event);
-        } catch (Exception e) {
             logger.warn("Exception occurred when storing new meta data", e);
         }
     }
 
+    @Override
+    public long getCurrentTerm() {
+        return previousManifest.getCurrentTerm();
+    }
+
+    @Override
+    public ClusterState getLastAcceptedState() {
+        assert previousClusterState.nodes().getLocalNode() != null : "Call setLocalNode before calling this method";
+        return previousClusterState;
+    }
+
+    @Override
+    public void setCurrentTerm(long currentTerm) {
+        Manifest manifest = new Manifest(currentTerm, previousManifest.getClusterStateVersion(), previousManifest.getGlobalGeneration(),
+                new HashMap<>(previousManifest.getIndexGenerations()));
+        try {
+            metaStateService.writeManifestAndCleanup("current term changed", manifest);
+            previousManifest = manifest;
+        } catch (WriteStateException e) {
+            logger.warn("Exception occurred when setting current term", e);
+            //TODO re-throw exception
+        }
+    }
+
+    @Override
+    public void setLastAcceptedState(ClusterState clusterState) {
+        assert clusterState.blocks().disableStatePersistence() == false;
+
+        try {
+            incrementalWrite = previousClusterState.term() == clusterState.term();
+            updateClusterState(clusterState, previousClusterState);
+        } catch (WriteStateException e) {
+            logger.warn("Exception occurred when setting last accepted state", e);
+            //TODO re-throw exception
+        }
+    }
+
     /**
      * This class is used to write changed global {@link MetaData}, {@link IndexMetaData} and {@link Manifest} to disk.
      * This class delegates <code>write*</code> calls to corresponding write calls in {@link MetaStateService} and
@@ -255,39 +301,41 @@ public class GatewayMetaState implements ClusterStateApplier {
     }
 
     /**
-     * Updates meta state and meta data on disk according to {@link ClusterChangedEvent}.
+     * Updates manifest and meta data on disk.
      *
-     * @throws IOException if IOException occurs. It's recommended for the callers of this method to handle {@link WriteStateException},
+     * @param newState new {@link ClusterState}
-     *                     which is subclass of {@link IOException} explicitly. See also {@link WriteStateException#isDirty()}.
+     * @param previousState previous {@link ClusterState}
+     *
+     * @throws WriteStateException if exception occurs. See also {@link WriteStateException#isDirty()}.
      */
-    private void updateMetaData(ClusterChangedEvent event) throws IOException {
+    protected void updateClusterState(ClusterState newState, ClusterState previousState)
-        ClusterState newState = event.state();
+            throws WriteStateException {
-        ClusterState previousState = event.previousState();
         MetaData newMetaData = newState.metaData();
 
         final AtomicClusterStateWriter writer = new AtomicClusterStateWriter(metaStateService, previousManifest);
         long globalStateGeneration = writeGlobalState(writer, newMetaData);
         Map<Index, Long> indexGenerations = writeIndicesMetadata(writer, newState, previousState);
-        Manifest manifest = new Manifest(globalStateGeneration, indexGenerations);
+        Manifest manifest = new Manifest(previousManifest.getCurrentTerm(), newState.version(), globalStateGeneration, indexGenerations);
         writeManifest(writer, manifest);
 
-        previousMetaData = newMetaData;
         previousManifest = manifest;
+        previousClusterState = newState;
     }
 
-    private void writeManifest(AtomicClusterStateWriter writer, Manifest manifest) throws IOException {
+    private void writeManifest(AtomicClusterStateWriter writer, Manifest manifest) throws WriteStateException {
         if (manifest.equals(previousManifest) == false) {
             writer.writeManifestAndCleanup("changed", manifest);
         }
     }
 
     private Map<Index, Long> writeIndicesMetadata(AtomicClusterStateWriter writer, ClusterState newState, ClusterState previousState)
-            throws IOException {
+            throws WriteStateException {
         Map<Index, Long> previouslyWrittenIndices = previousManifest.getIndexGenerations();
         Set<Index> relevantIndices = getRelevantIndices(newState, previousState, previouslyWrittenIndices.keySet());
 
         Map<Index, Long> newIndices = new HashMap<>();
 
+        MetaData previousMetaData = incrementalWrite ? previousState.metaData() : null;
         Iterable<IndexMetaDataAction> actions = resolveIndexMetaDataActions(previouslyWrittenIndices, relevantIndices, previousMetaData,
                 newState.metaData());
 
@@ -299,8 +347,9 @@ public class GatewayMetaState implements ClusterStateApplier {
         return newIndices;
     }
 
-    private long writeGlobalState(AtomicClusterStateWriter writer, MetaData newMetaData) throws IOException {
+    private long writeGlobalState(AtomicClusterStateWriter writer, MetaData newMetaData)
-        if (previousMetaData == null || MetaData.isGlobalStateEquals(previousMetaData, newMetaData) == false) {
+            throws WriteStateException {
+        if (incrementalWrite == false || MetaData.isGlobalStateEquals(previousClusterState.metaData(), newMetaData) == false) {
             return writer.writeGlobalState("changed", newMetaData);
         }
         return previousManifest.getGlobalGeneration();

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

@@ -94,7 +94,7 @@ public class GatewayService extends AbstractLifecycleComponent implements Cluste
 
     @Inject
     public GatewayService(Settings settings, AllocationService allocationService, ClusterService clusterService,
-                          ThreadPool threadPool, GatewayMetaState metaState,
+                          ThreadPool threadPool,
                           TransportNodesListGatewayMetaState listGatewayMetaState,
                           IndicesService indicesService) {
         super(settings);
@@ -124,8 +124,6 @@ public class GatewayService extends AbstractLifecycleComponent implements Cluste
             // TODO: change me once the minimum_master_nodes is changed too
             recoverAfterMasterNodes = settings.getAsInt("discovery.zen.minimum_master_nodes", -1);
         }
-
-        clusterService.addLowPriorityApplier(metaState);
     }
 
     @Override

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

@@ -138,7 +138,7 @@ public class MetaStateService {
             }
         }
 
-        Manifest manifest = new Manifest(globalStateGeneration, indices);
+        Manifest manifest = Manifest.unknownCurrentTermAndVersion(globalStateGeneration, indices);
         return new Tuple<>(manifest, metaDataBuilder.build());
     }
 
@@ -275,7 +275,7 @@ public class MetaStateService {
         Manifest manifest = loadManifestOrEmpty();
         Map<Index, Long> indices = new HashMap<>(manifest.getIndexGenerations());
         indices.put(metaData.getIndex(), generation);
-        manifest = new Manifest(manifest.getGlobalGeneration(), indices);
+        manifest = new Manifest(manifest.getCurrentTerm(), manifest.getClusterStateVersion(), manifest.getGlobalGeneration(), indices);
         writeManifestAndCleanup(reason, manifest);
         cleanupIndex(metaData.getIndex(), generation);
     }
@@ -287,7 +287,7 @@ public class MetaStateService {
     public void writeGlobalStateAndUpdateManifest(String reason, MetaData metaData) throws IOException {
         long generation = writeGlobalState(reason, metaData);
         Manifest manifest = loadManifestOrEmpty();
-        manifest = new Manifest(generation, manifest.getIndexGenerations());
+        manifest = new Manifest(manifest.getCurrentTerm(), manifest.getClusterStateVersion(), generation, manifest.getIndexGenerations());
         writeManifestAndCleanup(reason, manifest);
         cleanupGlobalState(generation);
     }

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

@@ -19,7 +19,6 @@
 
 package org.elasticsearch.gateway;
 
-import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.action.support.ActionFilters;
@@ -89,11 +88,7 @@ public class TransportNodesListGatewayMetaState extends TransportNodesAction<Tra
 
     @Override
     protected NodeGatewayMetaState nodeOperation(NodeRequest request) {
-        try {
+        return new NodeGatewayMetaState(clusterService.localNode(), metaState.getMetaData());
-            return new NodeGatewayMetaState(clusterService.localNode(), metaState.loadMetaData());
-        } catch (Exception e) {
-            throw new ElasticsearchException("failed to load metadata", e);
-        }
     }
 
     public static class Request extends BaseNodesRequest<Request> {

server/src/main/java/org/elasticsearch/node/Node.java

@@ -155,7 +155,6 @@ import javax.net.ssl.SNIHostName;
 import java.io.BufferedWriter;
 import java.io.Closeable;
 import java.io.IOException;
-import java.io.UncheckedIOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.nio.charset.Charset;
@@ -464,7 +463,7 @@ public class Node implements Closeable {
             final MetaDataIndexUpgradeService metaDataIndexUpgradeService = new MetaDataIndexUpgradeService(settings, xContentRegistry,
                 indicesModule.getMapperRegistry(), settingsModule.getIndexScopedSettings(), indexMetaDataUpgraders);
             final GatewayMetaState gatewayMetaState = new GatewayMetaState(settings, nodeEnvironment, metaStateService,
-                metaDataIndexUpgradeService, metaDataUpgrader);
+                    metaDataIndexUpgradeService, metaDataUpgrader);
             new TemplateUpgradeService(client, clusterService, threadPool, indexTemplateMetaDataUpgraders);
             final Transport transport = networkModule.getTransportSupplier().get();
             Set<String> taskHeaders = Stream.concat(
@@ -481,7 +480,7 @@ public class Node implements Closeable {
             final DiscoveryModule discoveryModule = new DiscoveryModule(this.settings, threadPool, transportService, namedWriteableRegistry,
                 networkService, clusterService.getMasterService(), clusterService.getClusterApplierService(),
                 clusterService.getClusterSettings(), pluginsService.filterPlugins(DiscoveryPlugin.class),
-                clusterModule.getAllocationService(), environment.configFile());
+                clusterModule.getAllocationService(), environment.configFile(), gatewayMetaState);
             this.nodeService = new NodeService(settings, threadPool, monitorService, discoveryModule.getDiscovery(),
                 transportService, indicesService, pluginsService, circuitBreakerService, scriptModule.getScriptService(),
                 httpServerTransport, ingestService, clusterService, settingsModule.getSettingsFilter(), responseCollectorService,
@@ -666,18 +665,14 @@ public class Node implements Closeable {
         assert transportService.getLocalNode().equals(localNodeFactory.getNode())
             : "transportService has a different local node than the factory provided";
         final MetaData onDiskMetadata;
-        try {
+        // we load the global state here (the persistent part of the cluster state stored on disk) to
-            // we load the global state here (the persistent part of the cluster state stored on disk) to
+        // pass it to the bootstrap checks to allow plugins to enforce certain preconditions based on the recovered state.
-            // pass it to the bootstrap checks to allow plugins to enforce certain preconditions based on the recovered state.
+        if (DiscoveryNode.isMasterNode(settings) || DiscoveryNode.isDataNode(settings)) {
-            if (DiscoveryNode.isMasterNode(settings) || DiscoveryNode.isDataNode(settings)) {
+            onDiskMetadata = injector.getInstance(GatewayMetaState.class).getMetaData();
-                onDiskMetadata = injector.getInstance(GatewayMetaState.class).loadMetaData();
+        } else {
-            } else {
+            onDiskMetadata = MetaData.EMPTY_META_DATA;
-                onDiskMetadata = MetaData.EMPTY_META_DATA;
-            }
-            assert onDiskMetadata != null : "metadata is null but shouldn't"; // this is never null
-        } catch (IOException e) {
-            throw new UncheckedIOException(e);
         }
+        assert onDiskMetadata != null : "metadata is null but shouldn't"; // this is never null
         validateNodeBeforeAcceptingRequests(new BootstrapContext(settings, onDiskMetadata), transportService.boundAddress(), pluginsService
             .filterPlugins(Plugin
             .class)

server/src/main/java/org/elasticsearch/plugins/DiscoveryPlugin.java

@@ -35,6 +35,7 @@ import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.discovery.Discovery;
 import org.elasticsearch.discovery.zen.UnicastHostsProvider;
+import org.elasticsearch.gateway.GatewayMetaState;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
@@ -73,7 +74,8 @@ public interface DiscoveryPlugin {
                                                                ClusterApplier clusterApplier,
                                                                ClusterSettings clusterSettings,
                                                                UnicastHostsProvider hostsProvider,
-                                                               AllocationService allocationService) {
+                                                               AllocationService allocationService,
+                                                               GatewayMetaState gatewayMetaState) {
         return Collections.emptyMap();
     }
 

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

@@ -0,0 +1,57 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.cluster.coordination;
+
+import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
+import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.equalTo;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0)
+public class PersistedStateIT extends ESIntegTestCase {
+
+    private static Settings SETTINGS = Settings.builder()
+        .put(TestZenDiscovery.USE_ZEN2.getKey(), true)
+        .put(TestZenDiscovery.USE_ZEN2_PERSISTED_STATE.getKey(), true)
+        .build();
+
+
+    public void testPersistentSettingsOnFullRestart() throws Exception {
+        internalCluster().startNodes(1, SETTINGS);
+        final int maxShardsPerNode = randomIntBetween(1000, 10000);
+
+        client().admin().cluster().updateSettings(
+                new ClusterUpdateSettingsRequest().persistentSettings(Settings.builder()
+                        .put(MetaData.SETTING_CLUSTER_MAX_SHARDS_PER_NODE.getKey(), maxShardsPerNode).build()))
+                .actionGet();
+        internalCluster().fullRestart();
+        ensureStableCluster(1);
+        MetaData metaData =
+                client().admin().cluster().state(new ClusterStateRequest()).actionGet(30, TimeUnit.SECONDS).getState().metaData();
+        assertThat(metaData.persistentSettings()
+                        .get(MetaData.SETTING_CLUSTER_MAX_SHARDS_PER_NODE.getKey()), equalTo(Integer.toString(maxShardsPerNode)));
+
+    }
+}

server/src/test/java/org/elasticsearch/cluster/metadata/ManifestTests.java

@@ -38,31 +38,46 @@ import static org.hamcrest.Matchers.equalTo;
 public class ManifestTests extends ESTestCase {
 
     private Manifest copyState(Manifest state, boolean introduceErrors) {
+        long currentTerm = state.getCurrentTerm();
+        long clusterStateVersion = state.getClusterStateVersion();
         long generation = state.getGlobalGeneration();
         Map<Index, Long> indices = new HashMap<>(state.getIndexGenerations());
         if (introduceErrors) {
             switch (randomInt(3)) {
                 case 0: {
-                    generation = generation + 1;
+                    currentTerm = randomValueOtherThan(currentTerm, () -> randomNonNegativeLong());
                     break;
                 }
                 case 1: {
-                    indices.remove(randomFrom(indices.keySet()));
+                    clusterStateVersion = randomValueOtherThan(clusterStateVersion, () -> randomNonNegativeLong());
                     break;
                 }
                 case 2: {
-                    Tuple<Index, Long> indexEntry = randomIndexEntry();
+                    generation = randomValueOtherThan(generation, () -> randomNonNegativeLong());
-                    indices.put(indexEntry.v1(), indexEntry.v2());
                     break;
                 }
                 case 3: {
-                    Index index = randomFrom(indices.keySet());
+                    switch (randomInt(2)) {
-                    indices.compute(index, (i, g) -> g + 1);
+                        case 0: {
+                            indices.remove(randomFrom(indices.keySet()));
+                            break;
+                        }
+                        case 1: {
+                            Tuple<Index, Long> indexEntry = randomIndexEntry();
+                            indices.put(indexEntry.v1(), indexEntry.v2());
+                            break;
+                        }
+                        case 2: {
+                            Index index = randomFrom(indices.keySet());
+                            indices.compute(index, (i, g) -> randomValueOtherThan(g, () -> randomNonNegativeLong()));
+                            break;
+                        }
+                    }
                     break;
                 }
             }
         }
-        return new Manifest(generation, indices);
+        return new Manifest(currentTerm, clusterStateVersion, generation, indices);
     }
 
     private Tuple<Index, Long> randomIndexEntry() {
@@ -74,13 +89,15 @@ public class ManifestTests extends ESTestCase {
     }
 
     private Manifest randomManifest() {
+        long currentTerm = randomNonNegativeLong();
+        long clusterStateVersion = randomNonNegativeLong();
         long generation = randomNonNegativeLong();
         Map<Index, Long> indices = new HashMap<>();
         for (int i = 0; i < randomIntBetween(1, 5); i++) {
             Tuple<Index, Long> indexEntry = randomIndexEntry();
             indices.put(indexEntry.v1(), indexEntry.v2());
         }
-        return new Manifest(generation, indices);
+        return new Manifest(currentTerm, clusterStateVersion, generation, indices);
     }
 
     public void testEqualsAndHashCode() {

server/src/test/java/org/elasticsearch/discovery/DiscoveryModuleTests.java

@@ -31,6 +31,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.discovery.zen.UnicastHostsProvider;
 import org.elasticsearch.discovery.zen.ZenDiscovery;
+import org.elasticsearch.gateway.GatewayMetaState;
 import org.elasticsearch.plugins.DiscoveryPlugin;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.NoopDiscovery;
@@ -60,6 +61,7 @@ public class DiscoveryModuleTests extends ESTestCase {
     private ClusterApplier clusterApplier;
     private ThreadPool threadPool;
     private ClusterSettings clusterSettings;
+    private GatewayMetaState gatewayMetaState;
 
     public interface DummyHostsProviderPlugin extends DiscoveryPlugin {
         Map<String, Supplier<UnicastHostsProvider>> impl();
@@ -77,7 +79,7 @@ public class DiscoveryModuleTests extends ESTestCase {
                                                                    NamedWriteableRegistry namedWriteableRegistry,
                                                                    MasterService masterService, ClusterApplier clusterApplier,
                                                                    ClusterSettings clusterSettings, UnicastHostsProvider hostsProvider,
-                                                                   AllocationService allocationService) {
+                                                                   AllocationService allocationService, GatewayMetaState gatewayMetaState) {
             return impl();
         }
     }
@@ -90,6 +92,7 @@ public class DiscoveryModuleTests extends ESTestCase {
         clusterApplier = mock(ClusterApplier.class);
         threadPool = mock(ThreadPool.class);
         clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
+        gatewayMetaState = mock(GatewayMetaState.class);
     }
 
     @After
@@ -99,7 +102,7 @@ public class DiscoveryModuleTests extends ESTestCase {
 
     private DiscoveryModule newModule(Settings settings, List<DiscoveryPlugin> plugins) {
         return new DiscoveryModule(settings, threadPool, transportService, namedWriteableRegistry, null, masterService,
-            clusterApplier, clusterSettings, plugins, null, createTempDir().toAbsolutePath());
+            clusterApplier, clusterSettings, plugins, null, createTempDir().toAbsolutePath(), gatewayMetaState);
     }
 
     public void testDefaults() {

server/src/test/java/org/elasticsearch/discovery/single/SingleNodeDiscoveryTests.java

@@ -69,7 +69,7 @@ public class SingleNodeDiscoveryTests extends ESTestCase {
                                 clusterState.set(clusterStateSupplier.get());
                                 listener.onSuccess(source);
                             }
-                        });
+                    });
             discovery.start();
             discovery.startInitialJoin();
             final DiscoveryNodes nodes = clusterState.get().nodes();

server/src/test/java/org/elasticsearch/discovery/zen/ZenDiscoveryUnitTests.java

@@ -50,6 +50,7 @@ import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
 import org.elasticsearch.discovery.zen.PublishClusterStateActionTests.AssertingAckListener;
 import org.elasticsearch.discovery.zen.ZenDiscovery.ZenNodeRemovalClusterStateTaskExecutor;
+import org.elasticsearch.gateway.GatewayMetaState;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.test.ClusterServiceUtils;
 import org.elasticsearch.test.ESTestCase;
@@ -362,7 +363,7 @@ public class ZenDiscoveryUnitTests extends ESTestCase {
             new NamedWriteableRegistry(ClusterModule.getNamedWriteables()),
             masterService, clusterApplier, clusterSettings, hostsResolver -> Collections.emptyList(),
             ESAllocationTestCase.createAllocationService(),
-            Collections.emptyList());
+            Collections.emptyList(), mock(GatewayMetaState.class));
         zenDiscovery.start();
         return zenDiscovery;
     }

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

@@ -0,0 +1,246 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.gateway;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.coordination.CoordinationMetaData;
+import org.elasticsearch.cluster.coordination.CoordinationMetaData.VotingTombstone;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.cluster.metadata.Manifest;
+import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.metadata.MetaDataIndexUpgradeService;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.plugins.MetaDataUpgrader;
+import org.elasticsearch.test.ESTestCase;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
+
+public class GatewayMetaStatePersistedStateTests extends ESTestCase {
+    private class GatewayMetaStateUT extends GatewayMetaState {
+        GatewayMetaStateUT(Settings settings, NodeEnvironment nodeEnvironment) throws IOException {
+            super(settings, nodeEnvironment, new MetaStateService(nodeEnvironment, xContentRegistry()),
+                    Mockito.mock(MetaDataIndexUpgradeService.class), Mockito.mock(MetaDataUpgrader.class));
+        }
+
+        @Override
+        protected void upgradeMetaData(MetaDataIndexUpgradeService metaDataIndexUpgradeService, MetaDataUpgrader metaDataUpgrader) {
+            // MetaData upgrade is tested in GatewayMetaStateTests, we override this method to NOP to make mocking easier
+        }
+    }
+
+    private NodeEnvironment nodeEnvironment;
+    private ClusterName clusterName;
+    private Settings settings;
+    private DiscoveryNode localNode;
+
+    @Override
+    public void setUp() throws Exception {
+        nodeEnvironment = newNodeEnvironment();
+        localNode = new DiscoveryNode("node1", buildNewFakeTransportAddress(), Collections.emptyMap(),
+                Sets.newHashSet(DiscoveryNode.Role.MASTER), Version.CURRENT);
+        clusterName = new ClusterName(randomAlphaOfLength(10));
+        settings = Settings.builder().put(ClusterName.CLUSTER_NAME_SETTING.getKey(), clusterName.value()).build();
+        super.setUp();
+    }
+
+    @Override
+    public void tearDown() throws Exception {
+        nodeEnvironment.close();
+        super.tearDown();
+    }
+
+    private GatewayMetaStateUT newGateway() throws IOException {
+        GatewayMetaStateUT gateway = new GatewayMetaStateUT(settings, nodeEnvironment);
+        gateway.setLocalNode(localNode);
+        return gateway;
+    }
+
+    private GatewayMetaStateUT maybeNew(GatewayMetaStateUT gateway) throws IOException {
+        if (randomBoolean()) {
+            return newGateway();
+        }
+        return gateway;
+    }
+
+    public void testInitialState() throws IOException {
+        GatewayMetaStateUT gateway = newGateway();
+        ClusterState state = gateway.getLastAcceptedState();
+        assertThat(state.getClusterName(), equalTo(clusterName));
+        assertTrue(MetaData.isGlobalStateEquals(state.metaData(), MetaData.EMPTY_META_DATA));
+        assertThat(state.getVersion(), equalTo(Manifest.empty().getClusterStateVersion()));
+        assertThat(state.getNodes().getLocalNode(), equalTo(localNode));
+
+        long currentTerm = gateway.getCurrentTerm();
+        assertThat(currentTerm, equalTo(Manifest.empty().getCurrentTerm()));
+    }
+
+    public void testSetCurrentTerm() throws IOException {
+        GatewayMetaStateUT gateway = newGateway();
+
+        for (int i = 0; i < randomIntBetween(1, 5); i++) {
+            final long currentTerm = randomNonNegativeLong();
+            gateway.setCurrentTerm(currentTerm);
+            gateway = maybeNew(gateway);
+            assertThat(gateway.getCurrentTerm(), equalTo(currentTerm));
+        }
+    }
+
+    private ClusterState createClusterState(long version, MetaData metaData) {
+        return ClusterState.builder(clusterName).
+                nodes(DiscoveryNodes.builder().add(localNode).localNodeId(localNode.getId()).build()).
+                version(version).
+                metaData(metaData).
+                build();
+    }
+
+    private CoordinationMetaData createCoordinationMetaData(long term) {
+        CoordinationMetaData.Builder builder = CoordinationMetaData.builder();
+        builder.term(term);
+        builder.lastAcceptedConfiguration(
+                new CoordinationMetaData.VotingConfiguration(
+                        Sets.newHashSet(generateRandomStringArray(10, 10, false))));
+        builder.lastCommittedConfiguration(
+                new CoordinationMetaData.VotingConfiguration(
+                        Sets.newHashSet(generateRandomStringArray(10, 10, false))));
+        for (int i = 0; i < randomIntBetween(0, 5); i++) {
+            builder.addVotingTombstone(new VotingTombstone(randomAlphaOfLength(10), randomAlphaOfLength(10)));
+        }
+
+        return builder.build();
+    }
+
+    private IndexMetaData createIndexMetaData(String indexName, int numberOfShards, long version) {
+        return IndexMetaData.builder(indexName).settings(
+                Settings.builder()
+                        .put(IndexMetaData.SETTING_INDEX_UUID, indexName)
+                        .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, numberOfShards)
+                        .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
+                        .put(IndexMetaData.SETTING_VERSION_CREATED, Version.CURRENT)
+                        .build()
+        ).version(version).build();
+    }
+
+    private void assertClusterStateEqual(ClusterState expected, ClusterState actual) {
+        assertThat(actual.version(), equalTo(expected.version()));
+        assertTrue(MetaData.isGlobalStateEquals(actual.metaData(), expected.metaData()));
+        for (IndexMetaData indexMetaData : expected.metaData()) {
+            assertThat(actual.metaData().index(indexMetaData.getIndex()), equalTo(indexMetaData));
+        }
+    }
+
+    public void testSetLastAcceptedState() throws IOException {
+        GatewayMetaStateUT gateway = newGateway();
+        final long term = randomNonNegativeLong();
+
+        for (int i = 0; i < randomIntBetween(1, 5); i++) {
+            final long version = randomNonNegativeLong();
+            final String indexName = randomAlphaOfLength(10);
+            final IndexMetaData indexMetaData = createIndexMetaData(indexName, randomIntBetween(1,5), randomNonNegativeLong());
+            final MetaData metaData = MetaData.builder().
+                    persistentSettings(Settings.builder().put(randomAlphaOfLength(10), randomAlphaOfLength(10)).build()).
+                    coordinationMetaData(createCoordinationMetaData(term)).
+                    put(indexMetaData, false).
+                    build();
+            ClusterState state = createClusterState(version, metaData);
+
+            gateway.setLastAcceptedState(state);
+            gateway = maybeNew(gateway);
+
+            ClusterState lastAcceptedState = gateway.getLastAcceptedState();
+            assertClusterStateEqual(state, lastAcceptedState);
+        }
+    }
+
+    public void testSetLastAcceptedStateTermChanged() throws IOException {
+        GatewayMetaStateUT gateway = newGateway();
+
+        final String indexName = randomAlphaOfLength(10);
+        final int numberOfShards = randomIntBetween(1, 5);
+        final long version = randomNonNegativeLong();
+        final long term = randomNonNegativeLong();
+        final IndexMetaData indexMetaData = createIndexMetaData(indexName, numberOfShards, version);
+        final ClusterState state = createClusterState(randomNonNegativeLong(),
+                MetaData.builder().coordinationMetaData(createCoordinationMetaData(term)).put(indexMetaData, false).build());
+        gateway.setLastAcceptedState(state);
+
+        gateway = maybeNew(gateway);
+        final long newTerm = randomValueOtherThan(term, () -> randomNonNegativeLong());
+        final int newNumberOfShards = randomValueOtherThan(numberOfShards, () -> randomIntBetween(1,5));
+        final IndexMetaData newIndexMetaData = createIndexMetaData(indexName, newNumberOfShards, version);
+        final ClusterState newClusterState = createClusterState(randomNonNegativeLong(),
+                MetaData.builder().coordinationMetaData(createCoordinationMetaData(newTerm)).put(newIndexMetaData, false).build());
+        gateway.setLastAcceptedState(newClusterState);
+
+        gateway = maybeNew(gateway);
+        assertThat(gateway.getLastAcceptedState().metaData().index(indexName), equalTo(newIndexMetaData));
+    }
+
+    public void testCurrentTermAndTermAreDifferent() throws IOException {
+        GatewayMetaStateUT gateway = newGateway();
+
+        long currentTerm = randomNonNegativeLong();
+        long term  = randomValueOtherThan(currentTerm, () -> randomNonNegativeLong());
+
+        gateway.setCurrentTerm(currentTerm);
+        gateway.setLastAcceptedState(createClusterState(randomNonNegativeLong(),
+                MetaData.builder().coordinationMetaData(CoordinationMetaData.builder().term(term).build()).build()));
+
+        gateway = maybeNew(gateway);
+        assertThat(gateway.getCurrentTerm(), equalTo(currentTerm));
+        assertThat(gateway.getLastAcceptedState().coordinationMetaData().term(), equalTo(term));
+    }
+
+    public void testMarkAcceptedConfigAsCommitted() throws IOException {
+        GatewayMetaStateUT gateway = newGateway();
+
+        CoordinationMetaData coordinationMetaData = createCoordinationMetaData(randomNonNegativeLong());
+        ClusterState state = createClusterState(randomNonNegativeLong(),
+                MetaData.builder().coordinationMetaData(coordinationMetaData).build());
+        gateway.setLastAcceptedState(state);
+
+        gateway = maybeNew(gateway);
+        assertThat(gateway.getLastAcceptedState().getLastAcceptedConfiguration(),
+                not(equalTo(gateway.getLastAcceptedState().getLastCommittedConfiguration())));
+        gateway.markLastAcceptedConfigAsCommitted();
+
+        CoordinationMetaData expectedCoordinationMetaData = CoordinationMetaData.builder(coordinationMetaData)
+                .lastCommittedConfiguration(coordinationMetaData.getLastAcceptedConfiguration()).build();
+        ClusterState expectedClusterState =
+                ClusterState.builder(state).metaData(MetaData.builder().coordinationMetaData(expectedCoordinationMetaData).build()).build();
+
+        gateway = maybeNew(gateway);
+        assertClusterStateEqual(expectedClusterState, gateway.getLastAcceptedState());
+        gateway.markLastAcceptedConfigAsCommitted();
+
+        gateway = maybeNew(gateway);
+        assertClusterStateEqual(expectedClusterState, gateway.getLastAcceptedState());
+    }
+}

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

@@ -383,6 +383,8 @@ public class GatewayMetaStateTests extends ESAllocationTestCase {
             Manifest manifest = Manifest.empty();
             MetaData metaData = MetaData.EMPTY_META_DATA;
             metaStateService.writeManifestAndCleanup("startup", Manifest.empty());
+            long currentTerm = randomNonNegativeLong();
+            long clusterStateVersion = randomNonNegativeLong();
 
             metaStateService.failRandomly();
             Set<MetaData> possibleMetaData = new HashSet<>();
@@ -402,7 +404,7 @@ public class GatewayMetaStateTests extends ESAllocationTestCase {
                         indexGenerations.put(indexMetaData.getIndex(), generation);
                     }
 
-                    Manifest newManifest = new Manifest(globalGeneration, indexGenerations);
+                    Manifest newManifest = new Manifest(currentTerm, clusterStateVersion, globalGeneration, indexGenerations);
                     writer.writeManifestAndCleanup("manifest", newManifest);
                     possibleMetaData.clear();
                     possibleMetaData.add(metaData);

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

@@ -35,7 +35,7 @@ public class GatewayServiceTests extends ESTestCase {
                 new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
                 null);
         return new GatewayService(settings.build(),
-                null, clusterService, null, null, null, null);
+                null, clusterService, null, null, null);
     }
 
     public void testDefaultRecoverAfterTime() throws IOException {

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

@@ -189,9 +189,9 @@ public class MetaDataWriteDataNodesIT extends ESIntegTestCase {
         return false;
     }
 
-    private ImmutableOpenMap<String, IndexMetaData> getIndicesMetaDataOnNode(String nodeName) throws Exception {
+    private ImmutableOpenMap<String, IndexMetaData> getIndicesMetaDataOnNode(String nodeName) {
         GatewayMetaState nodeMetaState = ((InternalTestCluster) cluster()).getInstance(GatewayMetaState.class, nodeName);
-        MetaData nodeMetaData = nodeMetaState.loadMetaData();
+        MetaData nodeMetaData = nodeMetaState.getMetaData();
         return nodeMetaData.getIndices();
     }
 }

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

@@ -140,9 +140,10 @@ public class MetaStateServiceTests extends ESTestCase {
     public void testLoadFullStateMissingGlobalMetaData() throws IOException {
         IndexMetaData index = indexMetaData("test1");
         long indexGeneration = metaStateService.writeIndex("test", index);
-        Manifest manifest = new Manifest(Manifest.empty().getGlobalGeneration(), new HashMap<Index, Long>() {{
+        Manifest manifest = new Manifest(randomNonNegativeLong(), randomNonNegativeLong(),
-            put(index.getIndex(), indexGeneration);
+                Manifest.empty().getGlobalGeneration(), new HashMap<Index, Long>() {{
-        }});
+                    put(index.getIndex(), indexGeneration);
+                }});
         assertTrue(manifest.isGlobalGenerationMissing());
         metaStateService.writeManifestAndCleanup("test", manifest);
 
@@ -164,10 +165,10 @@ public class MetaStateServiceTests extends ESTestCase {
         long globalGeneration = metaStateService.writeGlobalState("first global state write", metaData);
         long indexGeneration = metaStateService.writeIndex("first index state write", index);
 
-        Manifest manifest = new Manifest(globalGeneration, new HashMap<Index, Long>() {{
+        Manifest manifest = new Manifest(randomNonNegativeLong(), randomNonNegativeLong(),
+                globalGeneration, new HashMap<Index, Long>() {{
             put(index.getIndex(), indexGeneration);
         }});
-
         metaStateService.writeManifestAndCleanup("first manifest write", manifest);
 
         MetaData newMetaData = MetaData.builder()
@@ -184,7 +185,8 @@ public class MetaStateServiceTests extends ESTestCase {
         assertThat(loadedMetaData.hasIndex("test1"), equalTo(true));
         assertThat(loadedMetaData.index("test1"), equalTo(index));
 
-        manifest = new Manifest(globalGeneration, new HashMap<Index, Long>() {{
+        manifest = new Manifest(randomNonNegativeLong(), randomNonNegativeLong(),
+                globalGeneration, new HashMap<Index, Long>() {{
             put(index.getIndex(), indexGeneration);
         }});
 

server/src/test/java/org/elasticsearch/indices/IndicesServiceTests.java

@@ -232,12 +232,12 @@ public class IndicesServiceTests extends ESSingleNodeTestCase {
         }
 
         GatewayMetaState gwMetaState = getInstanceFromNode(GatewayMetaState.class);
-        MetaData meta = gwMetaState.loadMetaData();
+        MetaData meta = gwMetaState.getMetaData();
         assertNotNull(meta);
         assertNotNull(meta.index("test"));
         assertAcked(client().admin().indices().prepareDelete("test"));
 
-        meta = gwMetaState.loadMetaData();
+        meta = gwMetaState.getMetaData();
         assertNotNull(meta);
         assertNull(meta.index("test"));
 

test/framework/src/main/java/org/elasticsearch/test/discovery/TestZenDiscovery.java

@@ -27,6 +27,7 @@ import org.elasticsearch.cluster.coordination.InMemoryPersistedState;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.ClusterApplier;
+import org.elasticsearch.cluster.service.ClusterApplierService;
 import org.elasticsearch.cluster.service.MasterService;
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
@@ -38,6 +39,7 @@ import org.elasticsearch.discovery.DiscoveryModule;
 import org.elasticsearch.discovery.zen.UnicastHostsProvider;
 import org.elasticsearch.discovery.zen.ZenDiscovery;
 import org.elasticsearch.discovery.zen.ZenPing;
+import org.elasticsearch.gateway.GatewayMetaState;
 import org.elasticsearch.plugins.DiscoveryPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -64,6 +66,9 @@ public class TestZenDiscovery extends ZenDiscovery {
     public static final Setting<Boolean> USE_ZEN2 =
         Setting.boolSetting("discovery.zen.use_zen2", false, Setting.Property.NodeScope);
 
+    public static final Setting<Boolean> USE_ZEN2_PERSISTED_STATE =
+        Setting.boolSetting("discovery.zen.use_zen2_persisted_state", false, Setting.Property.NodeScope);
+
     /** A plugin which installs mock discovery and configures it to be used. */
     public static class TestPlugin extends Plugin implements DiscoveryPlugin {
         protected final Settings settings;
@@ -76,29 +81,43 @@ public class TestZenDiscovery extends ZenDiscovery {
                                                                   NamedWriteableRegistry namedWriteableRegistry,
                                                                   MasterService masterService, ClusterApplier clusterApplier,
                                                                   ClusterSettings clusterSettings, UnicastHostsProvider hostsProvider,
-                                                                  AllocationService allocationService) {
+                                                                  AllocationService allocationService, GatewayMetaState gatewayMetaState) {
             // we don't get the latest setting which were updated by the extra settings for the plugin. TODO: fix.
             Settings fixedSettings = Settings.builder().put(settings).putList(DISCOVERY_ZEN_PING_UNICAST_HOSTS_SETTING.getKey()).build();
             return Collections.singletonMap("test-zen", () -> {
                 if (USE_ZEN2.get(settings)) {
-                    // TODO: needs a proper storage layer
+                    Supplier<CoordinationState.PersistedState> persistedStateSupplier;
-                    Supplier<CoordinationState.PersistedState> persistedStateSupplier =
+                    if (USE_ZEN2_PERSISTED_STATE.get(settings)) {
-                        () -> new InMemoryPersistedState(0L, ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.get(settings))
+                        persistedStateSupplier = () -> {
-                            .nodes(DiscoveryNodes.builder().add(transportService.getLocalNode())
+                            gatewayMetaState.setLocalNode(transportService.getLocalNode());
-                                .localNodeId(transportService.getLocalNode().getId()).build()).build());
+                            return gatewayMetaState;
+                        };
+                    } else {
+                        if (clusterApplier instanceof ClusterApplierService) {
+                            //if InMemoryPersisted is used, we let GatewayMetaState receive all events,
+                            //because some tests rely on it due to dangling indices functionality.
+                            ((ClusterApplierService) clusterApplier).addLowPriorityApplier(gatewayMetaState);
+                        }
+
+                        persistedStateSupplier =
+                                () -> new InMemoryPersistedState(0L, ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.get(settings))
+                                        .nodes(DiscoveryNodes.builder().add(transportService.getLocalNode())
+                                                .localNodeId(transportService.getLocalNode().getId()).build()).build());
+                    }
+
                     return new Coordinator("test_node", fixedSettings, clusterSettings, transportService, namedWriteableRegistry,
                         allocationService, masterService, persistedStateSupplier, hostsProvider, clusterApplier,
                         new Random(Randomness.get().nextLong()));
                 } else {
                     return new TestZenDiscovery(fixedSettings, threadPool, transportService, namedWriteableRegistry, masterService,
-                        clusterApplier, clusterSettings, hostsProvider, allocationService);
+                        clusterApplier, clusterSettings, hostsProvider, allocationService, gatewayMetaState);
                 }
             });
         }
 
         @Override
         public List<Setting<?>> getSettings() {
-            return Arrays.asList(USE_MOCK_PINGS, USE_ZEN2);
+            return Arrays.asList(USE_MOCK_PINGS, USE_ZEN2, USE_ZEN2_PERSISTED_STATE);
         }
 
         @Override
@@ -113,9 +132,9 @@ public class TestZenDiscovery extends ZenDiscovery {
     private TestZenDiscovery(Settings settings, ThreadPool threadPool, TransportService transportService,
                              NamedWriteableRegistry namedWriteableRegistry, MasterService masterService,
                              ClusterApplier clusterApplier, ClusterSettings clusterSettings, UnicastHostsProvider hostsProvider,
-                             AllocationService allocationService) {
+                             AllocationService allocationService, GatewayMetaState gatewayMetaState) {
         super(settings, threadPool, transportService, namedWriteableRegistry, masterService, clusterApplier, clusterSettings,
-            hostsProvider, allocationService, Collections.emptyList());
+            hostsProvider, allocationService, Collections.emptyList(), gatewayMetaState);
     }
 
     @Override