[Zen2] PersistedState interface implementation

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

@@ -458,11 +458,13 @@ public interface PersistedState {
          */
         default void markLastAcceptedConfigAsCommitted() {
             final ClusterState lastAcceptedState = getLastAcceptedState();
-            final CoordinationMetaData coordinationMetaData = CoordinationMetaData.builder(lastAcceptedState.coordinationMetaData())
-                    .lastCommittedConfiguration(lastAcceptedState.getLastAcceptedConfiguration())
-                    .build();
-            final MetaData metaData = MetaData.builder(lastAcceptedState.metaData()).coordinationMetaData(coordinationMetaData).build();
-            setLastAcceptedState(ClusterState.builder(lastAcceptedState).metaData(metaData).build());
+            if (lastAcceptedState.getLastAcceptedConfiguration().equals(lastAcceptedState.getLastCommittedConfiguration()) == false) {
+                final CoordinationMetaData coordinationMetaData = CoordinationMetaData.builder(lastAcceptedState.coordinationMetaData())
+                        .lastCommittedConfiguration(lastAcceptedState.getLastAcceptedConfiguration())
+                        .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 @@
  * 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 Map<Index, Long> getIndexGenerations() {
         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 &&
-                Objects.equals(indexGenerations, manifest.indexGenerations);
+        return currentTerm == manifest.currentTerm &&
+               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 Manifest fromXContent(XContentParser parser) throws IOException {
      * 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) {
-        return (Long) generationAndListOfIndexEntries[0];
+    private static long currentTerm(Object[] manifestFields) {
+        return requireNonNullElseDefault((Long) manifestFields[0], MISSING_CURRENT_TERM);
+    }
+
+    private static long clusterStateVersion(Object[] manifestFields) {
+        return requireNonNullElseDefault((Long) manifestFields[1], MISSING_CLUSTER_STATE_VERSION);
+    }
+
+    private static long generation(Object[] manifestFields) {
+        return requireNonNullElseDefault((Long) manifestFields[2], MISSING_GLOBAL_GENERATION);
     }
 
-    private static Map<Index, Long> indices(Object[] generationAndListOfIndexEntries) {
-        List<IndexEntry> listOfIndices = (List<IndexEntry>) generationAndListOfIndexEntries[1];
+    @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 ->
-                    new Manifest(generation(generationAndListOfIndexEntries), indices(generationAndListOfIndexEntries)));
+            manifestFields ->
+                    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 static Manifest fromXContent(XContentParser parser) throws IOException {
     }
 
     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.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 DiscoveryModule(Settings settings, ThreadPool threadPool, TransportServic
         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 -> {
-                if (discoveryTypes.put(entry.getKey(), entry.getValue()) != null) {
-                    throw new IllegalArgumentException("Cannot register discovery type [" + entry.getKey() + "] twice");
-                }
-            });
+                masterService, clusterApplier, clusterSettings, hostsProvider, allocationService, gatewayMetaState).entrySet()
+                    .forEach(entry -> {
+                        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.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.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 ZenDiscovery(Settings settings, ThreadPool threadPool, TransportService t
 
         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(