Zen2: Add Cluster State Applier (#34257)

Adds the cluster state applier to Coordinator, and adds tests for cluster state acking.

Author: ywelsch

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

@@ -91,6 +91,10 @@ public void onFaultyNode(DiscoveryNode faultyNode) {
         onPossibleCompletion();
     }
 
+    public boolean isCommitted() {
+        return applyCommitRequest.isPresent();
+    }
+
     private void onPossibleCompletion() {
         if (isCompleted) {
             return;

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

@@ -20,7 +20,9 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportRequestOptions;
@@ -29,19 +31,20 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 import java.util.function.Function;
 
-public class PublicationTransportHandler {
+public class PublicationTransportHandler extends AbstractComponent {
 
     public static final String PUBLISH_STATE_ACTION_NAME = "internal:cluster/coordination/publish_state";
     public static final String COMMIT_STATE_ACTION_NAME = "internal:cluster/coordination/commit_state";
 
     private final TransportService transportService;
 
-    public PublicationTransportHandler(TransportService transportService,
+    public PublicationTransportHandler(Settings settings, TransportService transportService,
                                        Function<PublishRequest, PublishWithJoinResponse> handlePublishRequest,
-                                       Consumer<ApplyCommitRequest> handleApplyCommit) {
+                                       BiConsumer<ApplyCommitRequest, ActionListener<Void>> handleApplyCommit) {
+        super(settings);
         this.transportService = transportService;
 
         transportService.registerRequestHandler(PUBLISH_STATE_ACTION_NAME, ThreadPool.Names.GENERIC, false, false,
@@ -50,10 +53,27 @@ public PublicationTransportHandler(TransportService transportService,
 
         transportService.registerRequestHandler(COMMIT_STATE_ACTION_NAME, ThreadPool.Names.GENERIC, false, false,
             ApplyCommitRequest::new,
-            (request, channel, task) -> {
-                handleApplyCommit.accept(request);
-                channel.sendResponse(TransportResponse.Empty.INSTANCE);
-            });
+            (request, channel, task) -> handleApplyCommit.accept(request, new ActionListener<Void>() {
+
+                @Override
+                public void onResponse(Void aVoid) {
+                    try {
+                        channel.sendResponse(TransportResponse.Empty.INSTANCE);
+                    } catch (IOException e) {
+                        logger.debug("failed to send response on commit", e);
+                    }
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    try {
+                        channel.sendResponse(e);
+                    } catch (IOException ie) {
+                        e.addSuppressed(ie);
+                        logger.debug("failed to send response on commit", e);
+                    }
+                }
+            }));
     }
 
     public void sendPublishRequest(DiscoveryNode destination, PublishRequest publishRequest,