[Zen2] Move ClusterState fields to be persisted to ClusterState.MetaData (#35625)

Today we have a way to atomically persist global MetaData and
IndexMetaData to disk when new ClusterState is received. All other
ClusterState fields are not persisted.
However, there are other parts of ClusterState that should be
persisted, namely:

version
term
lastCommittedConfiguration
lastAcceptedConfiguration
votingTombstones

version is changed frequently, other fields are not. We decided
to group term, lastCommittedConfiguration,
lastAcceptedConfiguration and votingTombstones into
CoordinationMetaData class and make CoordinationMetaData a field
inside MetaData.
MetaData.toXContent and MetaData.fromXContent should take care of
CoordinationMetaData.
version stays as a top level field in ClusterState and will be
persisted as part of Manifest in a follow-up commit.
Also MetaData.isGlobalStateEquals should be extended to include
coordinationMetaData in comparison.

This commit favors exposing getters, such as getTerm directly in
ClusterState to avoid massive code changes.

An example of CoordinationMetaState.toXContent:

{
  "term": 1,
  "last_committed_config": [
    "TiIuBcbBtpuXyDDVHXeD",
    "ZIAoVbkjjLPLUuYLaTkw"
  ],
  "last_accepted_config": [
    "OwkXbXZNOZPJqccdFHdz",
    "LouzsGYwmQzpeQMrboZe",
    "fCKGRZdjLTqzXAqPUtGL",
    "pLoxshjpJXwDhbgjfYJy",
    "SjINLwFIlIEFZCbjrSFo",
    "MDkVncJEVyZLJktopWje"
  ]
}

Author: andrershov

server/src/main/java/org/elasticsearch/action/admin/cluster/bootstrap/BootstrapConfiguration.java

@@ -19,7 +19,7 @@
 package org.elasticsearch.action.admin.cluster.bootstrap;
 
 import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.cluster.ClusterState.VotingConfiguration;
+import org.elasticsearch.cluster.coordination.CoordinationMetaData.VotingConfiguration;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.io.stream.StreamInput;

server/src/main/java/org/elasticsearch/action/admin/cluster/configuration/TransportAddVotingTombstonesAction.java

@@ -24,13 +24,14 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterState.Builder;
 import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.ClusterStateObserver.Listener;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.coordination.CoordinationMetaData;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
@@ -90,9 +91,10 @@ public ClusterState execute(ClusterState currentState) {
                 assert resolvedNodes == null : resolvedNodes;
                 resolvedNodes = resolveNodesAndCheckMaximum(request, currentState);
 
-                final Builder builder = ClusterState.builder(currentState);
+                final CoordinationMetaData.Builder builder = CoordinationMetaData.builder(currentState.coordinationMetaData());
                 resolvedNodes.forEach(builder::addVotingTombstone);
-                final ClusterState newState = builder.build();
+                final MetaData newMetaData = MetaData.builder(currentState.metaData()).coordinationMetaData(builder.build()).build();
+                final ClusterState newState = ClusterState.builder(currentState).metaData(newMetaData).build();
                 assert newState.getVotingTombstones().size() <= MAXIMUM_VOTING_TOMBSTONES_SETTING.get(currentState.metaData().settings());
                 return newState;
             }

server/src/main/java/org/elasticsearch/action/admin/cluster/configuration/TransportClearVotingTombstonesAction.java

@@ -24,13 +24,14 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterState.Builder;
 import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.ClusterStateObserver.Listener;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.coordination.CoordinationMetaData;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
@@ -119,9 +120,11 @@ private void submitClearTombstonesTask(ClearVotingTombstonesRequest request, lon
         clusterService.submitStateUpdateTask("clear-voting-tombstones", new ClusterStateUpdateTask(Priority.URGENT) {
             @Override
             public ClusterState execute(ClusterState currentState) {
-                final Builder builder = ClusterState.builder(currentState);
-                builder.clearVotingTombstones();
-                return builder.build();
+                final CoordinationMetaData newCoordinationMetaData =
+                        CoordinationMetaData.builder(currentState.coordinationMetaData()).clearVotingTombstones().build();
+                final MetaData newMetaData = MetaData.builder(currentState.metaData()).
+                        coordinationMetaData(newCoordinationMetaData).build();
+                return ClusterState.builder(currentState).metaData(newMetaData).build();
             }
 
             @Override

server/src/main/java/org/elasticsearch/action/admin/cluster/state/TransportClusterStateAction.java

@@ -77,11 +77,9 @@ protected void masterOperation(final ClusterStateRequest request, final ClusterS
         ClusterState currentState = clusterService.state();
         logger.trace("Serving cluster state request using version {}", currentState.version());
         ClusterState.Builder builder = ClusterState.builder(currentState.getClusterName());
-        builder.term(currentState.term());
         builder.version(currentState.version());
         builder.stateUUID(currentState.stateUUID());
-        builder.lastCommittedConfiguration(currentState.getLastCommittedConfiguration());
-        builder.lastAcceptedConfiguration(currentState.getLastAcceptedConfiguration());
+
         if (request.nodes()) {
             builder.nodes(currentState.nodes());
         }
@@ -105,6 +103,7 @@ protected void masterOperation(final ClusterStateRequest request, final ClusterS
 
         MetaData.Builder mdBuilder = MetaData.builder();
         mdBuilder.clusterUUID(currentState.metaData().clusterUUID());
+        mdBuilder.coordinationMetaData(currentState.coordinationMetaData());
 
         if (request.metaData()) {
             if (request.indices().length > 0) {