[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/CoordinationState.java

@@ -406,6 +406,10 @@ public void invariant() {
         assert publishVotes.isEmpty() || electionWon();
     }
 
+    public boolean containsJoinVote(DiscoveryNode node) {
+        return joinVotes.containsVoteFor(node);
+    }
+
     /**
      * Pluggable persistence layer for {@link CoordinationState}.
      *

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,9 +28,14 @@
 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;
@@ -102,6 +108,35 @@ public String toString() {
             }));
     }
 
+    public void sendJoin(final DiscoveryNode destination, final JoinRequest joinRequest) {
+        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) {
+                final Throwable rootCause = exp.getRootCause();
+                if (rootCause instanceof CoordinationStateRejectedException) {
+                    logger.debug("failed to join {} with {}: {}", destination, joinRequest, rootCause.getMessage());
+                } else {
+                    logger.debug(() -> new ParameterizedMessage("failed to join {} with {}", destination, joinRequest), exp);
+                }
+            }
+
+            @Override
+            public String executor() {
+                return Names.SAME;
+            }
+        });
+    }
+
     public interface JoinCallback {
         void onSuccess();
 
@@ -211,7 +246,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

@@ -48,10 +48,9 @@ public class PreVoteCollector extends AbstractComponent {
     // Tuple for simple atomic updates
     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);
+        state = new Tuple<>(null, new PreVoteResponse(0, 0, 0));
         this.transportService = transportService;
         this.startElection = startElection;
 

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
+    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;
+                    }
                 });
             }
 
@@ -1176,7 +1181,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);
+                        }
+                    });
                 }
             }
         }