[Zen2] Add safety phase to CoordinatorTests

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

@@ -184,6 +184,14 @@ PublishWithJoinResponse handlePublishRequest(PublishRequest publishRequest) {
         synchronized (mutex) {
             final DiscoveryNode sourceNode = publishRequest.getAcceptedState().nodes().getMasterNode();
             logger.trace("handlePublishRequest: handling [{}] from [{}]", publishRequest, sourceNode);
+
+            if (sourceNode.equals(getLocalNode())
+                && (mode != Mode.LEADER || getCurrentTerm() != publishRequest.getAcceptedState().term())) {
+                // Rare case in which we stood down as leader between starting this publication and receiving it ourselves. The publication
+                // is already failed so there is no point in proceeding.
+                throw new CoordinationStateRejectedException("no longer leading this publication's term: " + publishRequest);
+            }
+
             ensureTermAtLeast(sourceNode, publishRequest.getAcceptedState().term());
             final PublishResponse publishResponse = coordinationState.get().handlePublishRequest(publishRequest);
 
@@ -600,7 +608,7 @@ protected void onCompletion(boolean committed) {
                         currentPublication = Optional.empty();
                         updateMaxTermSeen(getCurrentTerm()); // triggers term bump if new term was found during publication
 
-                        localNodeAckEvent.addListener(new ActionListener<Void>() {
+                        localNodeAckEvent.addListener(wrapWithMutex(new ActionListener<Void>() {
                             @Override
                             public void onResponse(Void ignore) {
                                 assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
@@ -628,7 +636,7 @@ public void onFailure(Exception e) {
                                 ackListener.onNodeAck(getLocalNode(), exception); // other nodes have acked, but not the master.
                                 publishListener.onFailure(exception);
                             }
-                        }, transportService.getThreadPool().generic());
+                        }), transportService.getThreadPool().generic());
                     }
 
                     @Override

server/src/main/java/org/elasticsearch/common/util/concurrent/ListenableFuture.java

@@ -82,7 +82,7 @@ protected synchronized void done() {
 
     private void notifyListener(ActionListener<V> listener, ExecutorService executorService) {
         try {
-            executorService.submit(new Runnable() {
+            executorService.execute(new Runnable() {
                 @Override
                 public void run() {
                     try {

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

@@ -36,6 +36,7 @@
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.threadpool.ThreadPool.Names;
 import org.elasticsearch.transport.TransportException;
+import org.elasticsearch.transport.TransportRequestOptions;
 import org.elasticsearch.transport.TransportResponseHandler;
 import org.elasticsearch.transport.TransportService;
 
@@ -59,7 +60,12 @@ public abstract class PeerFinder extends AbstractComponent {
         Setting.timeSetting("discovery.find_peers_interval",
             TimeValue.timeValueMillis(1000), TimeValue.timeValueMillis(1), Setting.Property.NodeScope);
 
-    private final TimeValue findPeersDelay;
+    public static final Setting<TimeValue> DISCOVERY_REQUEST_PEERS_TIMEOUT_SETTING =
+        Setting.timeSetting("discovery.request_peers_timeout",
+            TimeValue.timeValueMillis(10000), TimeValue.timeValueMillis(1), Setting.Property.NodeScope);
+
+    private final TimeValue findPeersInterval;
+    private final TimeValue requestPeersTimeout;
 
     private final Object mutex = new Object();
     private final TransportService transportService;
@@ -75,7 +81,8 @@ public abstract class PeerFinder extends AbstractComponent {
     public PeerFinder(Settings settings, TransportService transportService, TransportAddressConnector transportAddressConnector,
                       ConfiguredHostsResolver configuredHostsResolver) {
         super(settings);
-        findPeersDelay = DISCOVERY_FIND_PEERS_INTERVAL_SETTING.get(settings);
+        findPeersInterval = DISCOVERY_FIND_PEERS_INTERVAL_SETTING.get(settings);
+        requestPeersTimeout = DISCOVERY_REQUEST_PEERS_TIMEOUT_SETTING.get(settings);
         this.transportService = transportService;
         this.transportAddressConnector = transportAddressConnector;
         this.configuredHostsResolver = configuredHostsResolver;
@@ -241,7 +248,7 @@ private boolean handleWakeUp() {
             }
         });
 
-        transportService.getThreadPool().schedule(findPeersDelay, Names.GENERIC, new AbstractRunnable() {
+        transportService.getThreadPool().schedule(findPeersInterval, Names.GENERIC, new AbstractRunnable() {
             @Override
             public boolean isForceExecution() {
                 return true;
@@ -360,9 +367,11 @@ public void onFailure(Exception e) {
             });
         }
 
-        private void removePeer() {
+        void removePeer() {
             final Peer removed = peersByAddress.remove(transportAddress);
-            assert removed == Peer.this;
+            // assert removed == Peer.this : removed + " != " + Peer.this;
+            // ^ This assertion sometimes trips if we are deactivated and reactivated while a request is in flight.
+            // TODO be more careful about avoiding multiple active Peer objects for each address
         }
 
         private void requestPeers() {
@@ -380,6 +389,7 @@ private void requestPeers() {
 
             transportService.sendRequest(discoveryNode, REQUEST_PEERS_ACTION_NAME,
                 new PeersRequest(getLocalNode(), knownNodes),
+                TransportRequestOptions.builder().withTimeout(requestPeersTimeout).build(),
                 new TransportResponseHandler<PeersResponse>() {
 
                     @Override

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

@@ -47,13 +47,16 @@
 import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.transport.TransportService;
 import org.hamcrest.Matcher;
+import org.junit.Before;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.EnumSet;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Predicate;
@@ -65,6 +68,7 @@
 import static org.elasticsearch.cluster.coordination.CoordinationStateTests.value;
 import static org.elasticsearch.cluster.coordination.Coordinator.Mode.CANDIDATE;
 import static org.elasticsearch.cluster.coordination.Coordinator.Mode.FOLLOWER;
+import static org.elasticsearch.cluster.coordination.Coordinator.Mode.LEADER;
 import static org.elasticsearch.cluster.coordination.CoordinatorTests.Cluster.DEFAULT_DELAY_VARIABILITY;
 import static org.elasticsearch.cluster.coordination.ElectionSchedulerFactory.ELECTION_BACK_OFF_TIME_SETTING;
 import static org.elasticsearch.cluster.coordination.ElectionSchedulerFactory.ELECTION_INITIAL_TIMEOUT_SETTING;
@@ -86,8 +90,14 @@
 @TestLogging("org.elasticsearch.cluster.coordination:TRACE,org.elasticsearch.discovery:TRACE")
 public class CoordinatorTests extends ESTestCase {
 
+    @Before
+    public void resetPortCounterBeforeEachTest() {
+        resetPortCounter();
+    }
+
     public void testCanUpdateClusterStateAfterStabilisation() {
         final Cluster cluster = new Cluster(randomIntBetween(1, 5));
+        cluster.runRandomly();
         cluster.stabilise();
 
         final ClusterNode leader = cluster.getAnyLeader();
@@ -106,6 +116,7 @@ public void testCanUpdateClusterStateAfterStabilisation() {
 
     public void testNodesJoinAfterStableCluster() {
         final Cluster cluster = new Cluster(randomIntBetween(1, 5));
+        cluster.runRandomly();
         cluster.stabilise();
 
         final long currentTerm = cluster.getAnyLeader().coordinator.getCurrentTerm();
@@ -125,6 +136,7 @@ public void testNodesJoinAfterStableCluster() {
 
     public void testLeaderDisconnectionDetectedQuickly() {
         final Cluster cluster = new Cluster(randomIntBetween(3, 5));
+        cluster.runRandomly();
         cluster.stabilise();
 
         final ClusterNode originalLeader = cluster.getAnyLeader();
@@ -153,13 +165,14 @@ public void testLeaderDisconnectionDetectedQuickly() {
                 + DEFAULT_DELAY_VARIABILITY
                 // then wait for the removal to be committed
                 + DEFAULT_CLUSTER_STATE_UPDATE_DELAY
-            ));
+        ));
 
         assertThat(cluster.getAnyLeader().getId(), not(equalTo(originalLeader.getId())));
     }
 
     public void testUnresponsiveLeaderDetectedEventually() {
         final Cluster cluster = new Cluster(randomIntBetween(3, 5));
+        cluster.runRandomly();
         cluster.stabilise();
 
         final ClusterNode originalLeader = cluster.getAnyLeader();
@@ -189,6 +202,7 @@ public void testUnresponsiveLeaderDetectedEventually() {
 
     public void testFollowerDisconnectionDetectedQuickly() {
         final Cluster cluster = new Cluster(randomIntBetween(3, 5));
+        cluster.runRandomly();
         cluster.stabilise();
 
         final ClusterNode leader = cluster.getAnyLeader();
@@ -220,6 +234,7 @@ public void testFollowerDisconnectionDetectedQuickly() {
 
     public void testUnresponsiveFollowerDetectedEventually() {
         final Cluster cluster = new Cluster(randomIntBetween(3, 5));
+        cluster.runRandomly();
         cluster.stabilise();
 
         final ClusterNode leader = cluster.getAnyLeader();
@@ -289,6 +304,7 @@ private static String nodeIdFromIndex(int nodeIndex) {
 
     class Cluster {
 
+        static final long EXTREME_DELAY_VARIABILITY = 10000L;
         static final long DEFAULT_DELAY_VARIABILITY = 100L;
 
         final List<ClusterNode> clusterNodes;
@@ -299,6 +315,7 @@ class Cluster {
 
         private final Set<String> disconnectedNodes = new HashSet<>();
         private final Set<String> blackholedNodes = new HashSet<>();
+        private final Map<Long, ClusterState> committedStatesByVersion = new HashMap<>();
 
         Cluster(int initialNodeCount) {
             deterministicTaskQueue.setExecutionDelayVariabilityMillis(DEFAULT_DELAY_VARIABILITY);
@@ -328,6 +345,100 @@ void addNodes(int newNodesCount) {
             }
         }
 
+        void runRandomly() {
+
+            final int randomSteps = scaledRandomIntBetween(10, 10000);
+            logger.info("--> start of safety phase of at least [{}] steps", randomSteps);
+
+            deterministicTaskQueue.setExecutionDelayVariabilityMillis(EXTREME_DELAY_VARIABILITY);
+            int step = 0;
+            long finishTime = -1;
+
+            while (finishTime == -1 || deterministicTaskQueue.getCurrentTimeMillis() <= finishTime) {
+                step++;
+                if (randomSteps <= step && finishTime == -1) {
+                    finishTime = deterministicTaskQueue.getLatestDeferredExecutionTime();
+                    deterministicTaskQueue.setExecutionDelayVariabilityMillis(DEFAULT_DELAY_VARIABILITY);
+                    logger.debug("----> [runRandomly {}] reducing delay variability and running until [{}]", step, finishTime);
+                }
+
+                try {
+                    if (rarely()) {
+                        final ClusterNode clusterNode = getAnyNodePreferringLeaders();
+                        final int newValue = randomInt();
+                        logger.debug("----> [runRandomly {}] proposing new value [{}] to [{}]", step, newValue, clusterNode.getId());
+                        clusterNode.submitValue(newValue);
+                    } else if (rarely()) {
+                        final ClusterNode clusterNode = getAnyNode();
+                        logger.debug("----> [runRandomly {}] forcing {} to become candidate", step, clusterNode.getId());
+                        synchronized (clusterNode.coordinator.mutex) {
+                            clusterNode.coordinator.becomeCandidate("runRandomly");
+                        }
+                    } else if (rarely()) {
+                        final ClusterNode clusterNode = getAnyNode();
+                        final String id = clusterNode.getId();
+                        disconnectedNodes.remove(id);
+                        blackholedNodes.remove(id);
+
+                        switch (randomInt(2)) {
+                            case 0:
+                                logger.debug("----> [runRandomly {}] connecting {}", step, id);
+                                break;
+                            case 1:
+                                logger.debug("----> [runRandomly {}] disconnecting {}", step, id);
+                                disconnectedNodes.add(id);
+                                break;
+                            case 2:
+                                logger.debug("----> [runRandomly {}] blackholing {}", step, id);
+                                blackholedNodes.add(id);
+                                break;
+                        }
+                    } else {
+                        if (deterministicTaskQueue.hasDeferredTasks() && randomBoolean()) {
+                            deterministicTaskQueue.advanceTime();
+                        } else if (deterministicTaskQueue.hasRunnableTasks()) {
+                            deterministicTaskQueue.runRandomTask();
+                        }
+                    }
+
+                    // TODO other random steps:
+                    // - reboot a node
+                    // - abdicate leadership
+                    // - bootstrap
+
+                } catch (CoordinationStateRejectedException ignored) {
+                    // This is ok: it just means a message couldn't currently be handled.
+                }
+
+                assertConsistentStates();
+            }
+
+            disconnectedNodes.clear();
+            blackholedNodes.clear();
+        }
+
+        private void assertConsistentStates() {
+            for (final ClusterNode clusterNode : clusterNodes) {
+                clusterNode.coordinator.invariant();
+            }
+            updateCommittedStates();
+        }
+
+        private void updateCommittedStates() {
+            for (final ClusterNode clusterNode : clusterNodes) {
+                Optional<ClusterState> committedState = clusterNode.coordinator.getLastCommittedState();
+                if (committedState.isPresent()) {
+                    ClusterState storedState = committedStatesByVersion.get(committedState.get().getVersion());
+                    if (storedState == null) {
+                        committedStatesByVersion.put(committedState.get().getVersion(), committedState.get());
+                    } else {
+                        assertEquals("expected " + committedState.get() + " but got " + storedState,
+                            value(committedState.get()), value(storedState));
+                    }
+                }
+            }
+        }
+
         void stabilise() {
             stabilise(DEFAULT_STABILISATION_TIME);
         }
@@ -389,19 +500,20 @@ private void assertUniqueLeaderAndExpectedModes() {
 
                 final String nodeId = clusterNode.getId();
 
-                if (disconnectedNodes.contains(nodeId) || blackholedNodes.contains(nodeId)) {
-                    assertThat(nodeId + " is a candidate", clusterNode.coordinator.getMode(), is(CANDIDATE));
-                } else {
+                if (isConnectedPair(leader, clusterNode)) {
+                    assertThat(nodeId + " is a follower", clusterNode.coordinator.getMode(), is(FOLLOWER));
                     assertThat(nodeId + " has the same term as the leader", clusterNode.coordinator.getCurrentTerm(), is(leaderTerm));
-                    // TODO assert that all nodes have actually voted for the leader in this term
+                    // TODO assert that this node has actually voted for the leader in this term
+                    // TODO assert that this node's accepted and committed states are the same as the leader's
 
-                    assertThat(nodeId + " is a follower", clusterNode.coordinator.getMode(), is(FOLLOWER));
-                    assertThat(nodeId + " is at the same accepted version as the leader",
-                        Optional.of(clusterNode.coordinator.getLastAcceptedState().getVersion()), isPresentAndEqualToLeaderVersion);
-                    assertThat(nodeId + " is at the same committed version as the leader",
-                        clusterNode.coordinator.getLastCommittedState().map(ClusterState::getVersion), isPresentAndEqualToLeaderVersion);
-                    assertThat(clusterNode.coordinator.getLastCommittedState().map(ClusterState::getNodes).map(dn -> dn.nodeExists(nodeId)),
+                    assertThat(nodeId + " is in the leader's committed state",
+                        leader.coordinator.getLastCommittedState().map(ClusterState::getNodes).map(dn -> dn.nodeExists(nodeId)),
                         equalTo(Optional.of(true)));
+                } else {
+                    assertThat(nodeId + " is a candidate", clusterNode.coordinator.getMode(), is(CANDIDATE));
+                    assertThat(nodeId + " is not in the leader's committed state",
+                        leader.coordinator.getLastCommittedState().map(ClusterState::getNodes).map(dn -> dn.nodeExists(nodeId)),
+                        equalTo(Optional.of(false)));
                 }
             }
 
@@ -428,6 +540,10 @@ private ConnectionStatus getConnectionStatus(DiscoveryNode sender, DiscoveryNode
             return connectionStatus;
         }
 
+        ClusterNode getAnyNode() {
+            return getAnyNodeExcept();
+        }
+
         ClusterNode getAnyNodeExcept(ClusterNode... clusterNodes) {
             Set<String> forbiddenIds = Arrays.stream(clusterNodes).map(ClusterNode::getId).collect(Collectors.toSet());
             List<ClusterNode> acceptableNodes
@@ -436,6 +552,16 @@ ClusterNode getAnyNodeExcept(ClusterNode... clusterNodes) {
             return randomFrom(acceptableNodes);
         }
 
+        ClusterNode getAnyNodePreferringLeaders() {
+            for (int i = 0; i < 3; i++) {
+                ClusterNode clusterNode = getAnyNode();
+                if (clusterNode.coordinator.getMode() == LEADER) {
+                    return clusterNode;
+                }
+            }
+            return getAnyNode();
+        }
+
         class ClusterNode extends AbstractComponent {
             private final int nodeIndex;
             private Coordinator coordinator;
@@ -538,7 +664,7 @@ DiscoveryNode getLocalNode() {
             }
 
             boolean isLeader() {
-                return coordinator.getMode() == Coordinator.Mode.LEADER;
+                return coordinator.getMode() == LEADER;
             }
 
             void submitValue(final long value) {