Asyncify ElectionStrategy#beforeCommit (#94582)

Adjusts `ElectionStrategy#beforeCommit` to accept a listener, because
the stateless implementation will need to call into some async code.

Author: DaveCTurner

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

@@ -1690,8 +1690,8 @@ boolean cancelCommittedPublication() {
         }
     }
 
-    void beforeCommit(long term, long version) {
-        electionStrategy.beforeCommit(term, version);
+    private void beforeCommit(long term, long version, ActionListener<Void> listener) {
+        electionStrategy.beforeCommit(term, version, listener);
     }
 
     class CoordinatorPublication extends Publication {
@@ -1927,12 +1927,17 @@ protected boolean isPublishQuorum(VoteCollection votes) {
         }
 
         @Override
-        protected Optional<ApplyCommitRequest> handlePublishResponse(DiscoveryNode sourceNode, PublishResponse publishResponse) {
+        protected Optional<ListenableFuture<ApplyCommitRequest>> handlePublishResponse(
+            DiscoveryNode sourceNode,
+            PublishResponse publishResponse
+        ) {
             assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
             assert getCurrentTerm() >= publishResponse.getTerm();
-            var applyCommit = coordinationState.get().handlePublishResponse(sourceNode, publishResponse);
-            applyCommit.ifPresent(applyCommitRequest -> beforeCommit(applyCommitRequest.getTerm(), applyCommitRequest.getVersion()));
-            return applyCommit;
+            return coordinationState.get().handlePublishResponse(sourceNode, publishResponse).map(applyCommitRequest -> {
+                final var future = new ListenableFuture<ApplyCommitRequest>();
+                beforeCommit(applyCommitRequest.getTerm(), applyCommitRequest.getVersion(), future.map(ignored -> applyCommitRequest));
+                return future;
+            });
         }
 
         @Override
@@ -1981,6 +1986,7 @@ protected void sendApplyCommit(
             ActionListener<Empty> responseActionListener
         ) {
             assert transportService.getThreadPool().getThreadContext().isSystemContext();
+            assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
             try {
                 transportService.sendRequest(
                     destination,
@@ -1997,6 +2003,17 @@ protected void sendApplyCommit(
                 responseActionListener.onFailure(e);
             }
         }
+
+        @Override
+        protected <T> ActionListener<T> wrapListener(ActionListener<T> listener) {
+            return wrapWithMutex(listener);
+        }
+
+        @Override
+        boolean publicationCompletedIffAllTargetsInactiveOrCancelled() {
+            assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
+            return super.publicationCompletedIffAllTargetsInactiveOrCancelled();
+        }
     }
 
     public interface PeerFinderListener {

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

@@ -105,5 +105,7 @@ public boolean isInvalidReconfiguration(
             && lastCommittedConfiguration.equals(lastAcceptedConfiguration) == false;
     }
 
-    public void beforeCommit(long term, long version) {}
+    public void beforeCommit(long term, long version, ActionListener<Void> listener) {
+        listener.onResponse(null);
+    }
 }

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

@@ -16,6 +16,7 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.coordination.ClusterStatePublisher.AckListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportResponse;
@@ -37,7 +38,7 @@ public abstract class Publication {
     private final LongSupplier currentTimeSupplier;
     private final long startTime;
 
-    private Optional<ApplyCommitRequest> applyCommitRequest; // set when state is committed
+    private Optional<ListenableFuture<ApplyCommitRequest>> applyCommitRequest; // set when state is committed
     private boolean isCompleted; // set when publication is completed
     private boolean cancelled; // set when publication is cancelled
 
@@ -122,7 +123,7 @@ private void onPossibleCompletion() {
     }
 
     // For assertions only: verify that this invariant holds
-    private boolean publicationCompletedIffAllTargetsInactiveOrCancelled() {
+    boolean publicationCompletedIffAllTargetsInactiveOrCancelled() {
         if (cancelled == false) {
             for (final PublicationTarget target : publicationTargets) {
                 if (target.isActive()) {
@@ -173,7 +174,10 @@ protected final long getStartTime() {
 
     protected abstract boolean isPublishQuorum(CoordinationState.VoteCollection votes);
 
-    protected abstract Optional<ApplyCommitRequest> handlePublishResponse(DiscoveryNode sourceNode, PublishResponse publishResponse);
+    protected abstract Optional<ListenableFuture<ApplyCommitRequest>> handlePublishResponse(
+        DiscoveryNode sourceNode,
+        PublishResponse publishResponse
+    );
 
     protected abstract void onJoin(Join join);
 
@@ -191,6 +195,8 @@ protected abstract void sendApplyCommit(
         ActionListener<TransportResponse.Empty> responseActionListener
     );
 
+    protected abstract <T> ActionListener<T> wrapListener(ActionListener<T> listener);
+
     @Override
     public String toString() {
         return "Publication{term="
@@ -280,7 +286,19 @@ void sendApplyCommit() {
             assert state == PublicationTargetState.WAITING_FOR_QUORUM : state + " -> " + PublicationTargetState.SENT_APPLY_COMMIT;
             state = PublicationTargetState.SENT_APPLY_COMMIT;
             assert applyCommitRequest.isPresent();
-            Publication.this.sendApplyCommit(discoveryNode, applyCommitRequest.get(), new ApplyCommitResponseHandler());
+            applyCommitRequest.get().addListener(wrapListener(new ActionListener<>() {
+                @Override
+                public void onResponse(ApplyCommitRequest applyCommitRequest) {
+                    Publication.this.sendApplyCommit(discoveryNode, applyCommitRequest, new ApplyCommitResponseHandler());
+                    assert publicationCompletedIffAllTargetsInactiveOrCancelled();
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    setFailed(e);
+                    onPossibleCommitFailure();
+                }
+            }));
             assert publicationCompletedIffAllTargetsInactiveOrCancelled();
         }
 

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

@@ -515,11 +515,13 @@ public boolean isInvalidReconfiguration(
         }
 
         @Override
-        public void beforeCommit(long term, long version) {
+        public void beforeCommit(long term, long version, ActionListener<Void> listener) {
             // TODO: add a test to ensure that this gets called
             final var currentTermOwner = register.getTermOwner();
             if (currentTermOwner.term() > term) {
-                throw new CoordinationStateRejectedException("Term " + term + " already claimed by another node");
+                listener.onFailure(new CoordinationStateRejectedException("Term " + term + " already claimed by another node"));
+            } else {
+                listener.onResponse(null);
             }
         }
     }

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

@@ -16,6 +16,7 @@
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
@@ -90,8 +91,20 @@ protected boolean isPublishQuorum(CoordinationState.VoteCollection votes) {
                 }
 
                 @Override
-                protected Optional<ApplyCommitRequest> handlePublishResponse(DiscoveryNode sourceNode, PublishResponse publishResponse) {
-                    return coordinationState.handlePublishResponse(sourceNode, publishResponse);
+                protected Optional<ListenableFuture<ApplyCommitRequest>> handlePublishResponse(
+                    DiscoveryNode sourceNode,
+                    PublishResponse publishResponse
+                ) {
+                    return coordinationState.handlePublishResponse(sourceNode, publishResponse).map(applyCommitRequest -> {
+                        final var future = new ListenableFuture<ApplyCommitRequest>();
+                        future.onResponse(applyCommitRequest);
+                        return future;
+                    });
+                }
+
+                @Override
+                protected <T> ActionListener<T> wrapListener(ActionListener<T> listener) {
+                    return listener;
                 }
             };
             currentPublication.start(faultyNodes);