[Zen2] Implement basic cluster formation

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

@@ -49,6 +49,7 @@ public String toString() {
         return "ApplyCommitRequest{" +
             "term=" + term +
             ", version=" + version +
+            ", sourceNode=" + sourceNode +
             '}';
     }
 }

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

@@ -18,6 +18,7 @@
  */
 package org.elasticsearch.cluster.coordination;
 
+import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateTaskConfig;
 import org.elasticsearch.cluster.ClusterStateTaskListener;
@@ -27,29 +28,37 @@
 import org.elasticsearch.cluster.service.MasterService;
 import org.elasticsearch.common.Priority;
 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.threadpool.ThreadPool.Names;
+import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportResponse;
+import org.elasticsearch.transport.TransportResponse.Empty;
+import org.elasticsearch.transport.TransportResponseHandler;
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.BiConsumer;
+import java.util.function.Function;
 import java.util.function.LongSupplier;
 
 public class JoinHelper extends AbstractComponent {
 
     public static final String JOIN_ACTION_NAME = "internal:cluster/coordination/join";
+    public static final String START_JOIN_ACTION_NAME = "internal:cluster/coordination/start_join";
 
     private final MasterService masterService;
     private final TransportService transportService;
     private final JoinTaskExecutor joinTaskExecutor;
 
     public JoinHelper(Settings settings, AllocationService allocationService, MasterService masterService,
                       TransportService transportService, LongSupplier currentTermSupplier,
-                      BiConsumer<JoinRequest, JoinCallback> joinHandler) {
+                      BiConsumer<JoinRequest, JoinCallback> joinHandler, Function<StartJoinRequest, Join> joinLeaderInTerm) {
         super(settings);
         this.masterService = masterService;
         this.transportService = transportService;
@@ -100,6 +109,62 @@ public String toString() {
                     return "JoinCallback{request=" + request + "}";
                 }
             }));
+
+        transportService.registerRequestHandler(START_JOIN_ACTION_NAME, Names.GENERIC, false, false,
+            StartJoinRequest::new,
+            (request, channel, task) -> {
+                final DiscoveryNode destination = request.getSourceNode();
+                final JoinRequest joinRequest
+                    = new JoinRequest(transportService.getLocalNode(), Optional.of(joinLeaderInTerm.apply(request)));
+                logger.debug("attempting to join {} with {}", destination, joinRequest);
+                this.transportService.sendRequest(destination, JOIN_ACTION_NAME, joinRequest, new TransportResponseHandler<Empty>() {
+                    @Override
+                    public Empty read(StreamInput in) {
+                        return Empty.INSTANCE;
+                    }
+
+                    @Override
+                    public void handleResponse(Empty response) {
+                        logger.debug("successfully joined {} with {}", destination, joinRequest);
+                    }
+
+                    @Override
+                    public void handleException(TransportException exp) {
+                        logger.debug(() -> new ParameterizedMessage("failed to join {} with {}", destination, joinRequest), exp);
+                    }
+
+                    @Override
+                    public String executor() {
+                        return Names.SAME;
+                    }
+                });
+                channel.sendResponse(Empty.INSTANCE);
+            });
+    }
+
+    public void sendStartJoinRequest(final StartJoinRequest startJoinRequest, final DiscoveryNode destination) {
+        transportService.sendRequest(destination, START_JOIN_ACTION_NAME,
+            startJoinRequest, new TransportResponseHandler<Empty>() {
+                @Override
+                public Empty read(StreamInput in) {
+                    return Empty.INSTANCE;
+                }
+
+                @Override
+                public void handleResponse(Empty response) {
+                    logger.debug("successful response to {} from {}", startJoinRequest, destination);
+                }
+
+                @Override
+                public void handleException(TransportException exp) {
+                    logger.debug(new ParameterizedMessage("failure in response to {} from {}", startJoinRequest, destination), exp);
+                }
+
+                @Override
+                public String executor() {
+                    return ThreadPool.Names.SAME;
+                }
+            });
     }
 
     public interface JoinCallback {
@@ -211,7 +276,8 @@ public void close(Mode newMode) {
 
         @Override
         public String toString() {
-            return "CandidateJoinAccumulator{" + joinRequestAccumulator.keySet() + '}';
+            return "CandidateJoinAccumulator{" + joinRequestAccumulator.keySet() +
+                ", closed=" + closed + '}';
         }
     }
 }

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

@@ -23,6 +23,7 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.coordination.CoordinationState.VoteCollection;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.lease.Releasable;
@@ -45,13 +46,11 @@ public class PreVoteCollector extends AbstractComponent {
     private final TransportService transportService;
     private final Runnable startElection;
 
-    // Tuple for simple atomic updates
-    private volatile Tuple<DiscoveryNode, PreVoteResponse> state; // DiscoveryNode component is null if there is currently no known leader
+    // Tuple for simple atomic updates. null until the first call to `update()`.
+    private volatile Tuple<DiscoveryNode, PreVoteResponse> state; // DiscoveryNode component is null if there is currently no known leader.
 
-    PreVoteCollector(final Settings settings, final PreVoteResponse preVoteResponse,
-                     final TransportService transportService, final Runnable startElection) {
+    PreVoteCollector(final Settings settings, final TransportService transportService, final Runnable startElection) {
         super(settings);
-        state = new Tuple<>(null, preVoteResponse);
         this.transportService = transportService;
         this.startElection = startElection;
 
@@ -74,7 +73,7 @@ public Releasable start(final ClusterState clusterState, final Iterable<Discover
         return preVotingRound;
     }
 
-    public void update(final PreVoteResponse preVoteResponse, final DiscoveryNode leader) {
+    public void update(final PreVoteResponse preVoteResponse, @Nullable final DiscoveryNode leader) {
         logger.trace("updating with preVoteResponse={}, leader={}", preVoteResponse, leader);
         state = new Tuple<>(leader, preVoteResponse);
     }

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

@@ -156,7 +156,7 @@ private void onPossibleCommitFailure() {
 
     protected abstract Optional<ApplyCommitRequest> handlePublishResponse(DiscoveryNode sourceNode, PublishResponse publishResponse);
 
-    protected abstract void onPossibleJoin(DiscoveryNode sourceNode, PublishWithJoinResponse response);
+    protected abstract void onJoin(Join join);
 
     protected abstract void sendPublishRequest(DiscoveryNode destination, PublishRequest publishRequest,
                                                ActionListener<PublishWithJoinResponse> responseActionListener);
@@ -287,8 +287,10 @@ public void onResponse(PublishWithJoinResponse response) {
                     return;
                 }
 
-                // TODO: check if we need to pass the full response here or if it's sufficient to just pass the optional join.
-                onPossibleJoin(discoveryNode, response);
+                response.getJoin().ifPresent(join -> {
+                    assert discoveryNode.equals(join.getSourceNode());
+                    onJoin(join);
+                });
 
                 assert state == PublicationTargetState.SENT_PUBLISH_REQUEST : state + " -> " + PublicationTargetState.WAITING_FOR_QUORUM;
                 state = PublicationTargetState.WAITING_FOR_QUORUM;

server/src/main/java/org/elasticsearch/discovery/PeerFinder.java

@@ -46,6 +46,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Consumer;
+import java.util.function.LongConsumer;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.common.util.concurrent.ConcurrentCollections.newConcurrentMap;
@@ -65,20 +66,22 @@ public abstract class PeerFinder extends AbstractComponent {
     private final TransportService transportService;
     private final TransportAddressConnector transportAddressConnector;
     private final ConfiguredHostsResolver configuredHostsResolver;
+    private final LongConsumer updateMaxTermSeen;
 
     private volatile long currentTerm;
     private boolean active;
     private DiscoveryNodes lastAcceptedNodes;
     private final Map<TransportAddress, Peer> peersByAddress = newConcurrentMap();
     private Optional<DiscoveryNode> leader = Optional.empty();
 
-    PeerFinder(Settings settings, TransportService transportService,
-               TransportAddressConnector transportAddressConnector, ConfiguredHostsResolver configuredHostsResolver) {
+    public PeerFinder(Settings settings, TransportService transportService, TransportAddressConnector transportAddressConnector,
+                      ConfiguredHostsResolver configuredHostsResolver, LongConsumer updateMaxTermSeen) {
         super(settings);
         findPeersDelay = DISCOVERY_FIND_PEERS_INTERVAL_SETTING.get(settings);
         this.transportService = transportService;
         this.transportAddressConnector = transportAddressConnector;
         this.configuredHostsResolver = configuredHostsResolver;
+        this.updateMaxTermSeen = updateMaxTermSeen;
 
         transportService.registerRequestHandler(REQUEST_PEERS_ACTION_NAME, Names.GENERIC, false, false,
             PeersRequest::new,
@@ -95,6 +98,8 @@ public void activate(final DiscoveryNodes lastAcceptedNodes) {
             leader = Optional.empty();
             handleWakeUp(); // return value discarded: there are no known peers, so none can be disconnected
         }
+
+        onFoundPeersUpdated(); // trigger a check for a quorum already
     }
 
     public void deactivate(DiscoveryNode leader) {
@@ -160,6 +165,11 @@ private DiscoveryNode getLocalNode() {
      */
     protected abstract void onFoundPeersUpdated();
 
+    // only for assertions, but accessed from separate package o.e.c.coordination so needs to be public
+    public boolean isActive() {
+        return active;
+    }
+
     public interface TransportAddressConnector {
         /**
          * Identify the node at the given address and, if it is a master node and not the local node then establish a full connection to it.
@@ -247,7 +257,7 @@ protected void doRun() {
 
             @Override
             public String toString() {
-                return "PeerFinder::handleWakeUp";
+                return "PeerFinder handling wakeup";
             }
         });
 

server/src/main/java/org/elasticsearch/transport/TransportService.java

@@ -733,6 +733,11 @@ public void onFailure(Exception e) {
                                     "failed to notify channel of error message for action [{}]", action), inner);
                         }
                     }
+
+                    @Override
+                    public String toString() {
+                        return "processing of [" + action + "][" + requestId + "]: " + request;
+                    }
                 });
             }
 
@@ -1049,6 +1054,11 @@ public void cancel() {
                 "cancel must be called after the requestId [" + requestId + "] has been removed from clientHandlers";
             FutureUtils.cancel(future);
         }
+
+        @Override
+        public String toString() {
+            return "TimeoutHandler for [" + action + "][" + requestId + "]";
+        }
     }
 
     static class TimeoutInfoHolder {
@@ -1176,7 +1186,17 @@ public void sendResponse(final TransportResponse response, TransportResponseOpti
                 if (ThreadPool.Names.SAME.equals(executor)) {
                     processResponse(handler, response);
                 } else {
-                    threadPool.executor(executor).execute(() -> processResponse(handler, response));
+                    threadPool.executor(executor).execute(new Runnable() {
+                        @Override
+                        public String toString() {
+                            return "delivery of response to [" + action + "][" + requestId + "]: " + response;
+                        }
+
+                        @Override
+                        public void run() {
+                            DirectResponseChannel.this.processResponse(handler, response);
+                        }
+                    });
                 }
             }
         }