Add core coordination algorithm for cluster state publishing

build.gradle

@@ -173,8 +173,8 @@ task verifyVersions {
  * the enabled state of every bwc task. It should be set back to true
  * after the backport of the backcompat code is complete.
  */
-final boolean bwc_tests_enabled = true
-final String bwc_tests_disabled_issue = "" /* place a PR link here when committing bwc changes */
+final boolean bwc_tests_enabled = false
+final String bwc_tests_disabled_issue = "https://github.com/elastic/elasticsearch/issues/32006" // BWC handled at a later time
 if (bwc_tests_enabled == false) {
   if (bwc_tests_disabled_issue.isEmpty()) {
     throw new GradleException("bwc_tests_disabled_issue must be set when bwc_tests_enabled == false")

server/src/main/java/org/elasticsearch/ElasticsearchException.java

@@ -21,6 +21,7 @@
 
 import org.elasticsearch.action.support.replication.ReplicationOperation;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
+import org.elasticsearch.cluster.coordination.CoordinationStateRejectedException;
 import org.elasticsearch.common.CheckedFunction;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.ParseField;
@@ -1024,8 +1025,10 @@ private enum ElasticsearchExceptionHandle {
         UNKNOWN_NAMED_OBJECT_EXCEPTION(org.elasticsearch.common.xcontent.UnknownNamedObjectException.class,
                 org.elasticsearch.common.xcontent.UnknownNamedObjectException::new, 148, Version.V_5_2_0),
         TOO_MANY_BUCKETS_EXCEPTION(MultiBucketConsumerService.TooManyBucketsException.class,
-                                   MultiBucketConsumerService.TooManyBucketsException::new, 149,
-            Version.V_7_0_0_alpha1);
+            MultiBucketConsumerService.TooManyBucketsException::new, 149, Version.V_7_0_0_alpha1),
+        COORDINATION_STATE_REJECTED_EXCEPTION(CoordinationStateRejectedException.class,
+            CoordinationStateRejectedException::new, 150, Version.V_7_0_0_alpha1);
+
 
         final Class<? extends ElasticsearchException> exceptionClass;
         final CheckedFunction<StreamInput, ? extends ElasticsearchException, IOException> constructor;

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

@@ -76,8 +76,11 @@ 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());
         }