cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aweisb...@apache.org
Subject cassandra git commit: Startup checker should wait for count rather than percentage
Date Mon, 12 Nov 2018 17:45:35 GMT
Repository: cassandra
Updated Branches:
  refs/heads/trunk 918b1d8c6 -> 801cb70ee


Startup checker should wait for count rather than percentage

This improves on the wait for healthy work from CASSANDRA-13993 to
solve CASSANDRA-14297. In particular now the connectivity checker waits
for all but a single node in either the local datacenter or every
datacenter (defaults to just local, but the user can configure it to
wait for every datacenter). This way users can use this feature to ensure
availability of their application during restarts of Cassandra. The default
behavior waits for all but a single local datacenter node.

Patch by Joseph Lynch; Reviewed by Ariel Weisberg for CASSANDRA-14297


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/801cb70e
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/801cb70e
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/801cb70e

Branch: refs/heads/trunk
Commit: 801cb70ee811c956e987718a00695638d5bec1b6
Parents: 918b1d8
Author: Joseph Lynch <joe.e.lynch@gmail.com>
Authored: Thu Aug 23 15:19:20 2018 -0700
Committer: Ariel Weisberg <aweisberg@apple.com>
Committed: Mon Nov 12 12:41:17 2018 -0500

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 NEWS.txt                                        |   7 +
 .../org/apache/cassandra/config/Config.java     |  23 ++-
 .../cassandra/config/DatabaseDescriptor.java    |   4 +-
 .../net/StartupClusterConnectivityChecker.java  | 137 ++++++++++----
 .../cassandra/service/CassandraDaemon.java      |   6 +-
 .../StartupClusterConnectivityCheckerTest.java  | 179 +++++++++++++++++--
 7 files changed, 299 insertions(+), 58 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/801cb70e/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index a7a75c0..aaea773 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 4.0
+ * Startup checker should wait for count rather than percentage (CASSANDRA-14297)
  * Fix incorrect sorting of replicas in SimpleStrategy.calculateNaturalReplicas (CASSANDRA-14862)
  * Partitioned outbound internode TCP connections can occur when nodes restart (CASSANDRA-14358)
  * Don't write to system_distributed.repair_history, system_traces.sessions, system_traces.events
in mixed version 3.X/4.0 clusters (CASSANDRA-14841)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/801cb70e/NEWS.txt
----------------------------------------------------------------------
diff --git a/NEWS.txt b/NEWS.txt
index 0d211a3..63c4a47 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -38,6 +38,13 @@ using the provided 'sstableupgrade' tool.
 
 New features
 ------------
+   - Nodes will now bootstrap all intra-cluster connections at startup by default and wait
+     10 seconds for the all but one node in the local data center to be connected and marked
+     UP in gossip. This prevents nodes from coordinating requests and failing because they
+     aren't able to connect to the cluster fast enough. block_for_peers_timeout_in_secs in
+     cassandra.yaml can be used to configure how long to wait (or whether to wait at all)
+     and block_for_peers_in_remote_dcs can be used to also block on all but one node in
+     each remote DC as well. See CASSANDRA-14297 and CASSANDRA-13993 for more information.
    - *Experimental* support for Transient Replication and Cheap Quorums introduced by CASSANDRA-14404
      The intended audience for this functionality is expert users of Cassandra who are prepared
      to validate every aspect of the database for their application and deployment practices.
Future

http://git-wip-us.apache.org/repos/asf/cassandra/blob/801cb70e/src/java/org/apache/cassandra/config/Config.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index 1e80108..7371df7 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -388,9 +388,28 @@ public class Config
     public RepairCommandPoolFullStrategy repair_command_pool_full_strategy = RepairCommandPoolFullStrategy.queue;
     public int repair_command_pool_size = concurrent_validations;
 
-    // parameters to adjust how much to delay startup until a certain amount of the cluster
is connect to and marked alive
-    public int block_for_peers_percentage = 70;
+    /**
+     * When a node first starts up it intially considers all other peers as DOWN and is disconnected
from all of them.
+     * To be useful as a coordinator (and not introduce latency penalties on restart) this
node must have successfully
+     * opened all three internode TCP connections (gossip, small, and large messages) before
advertising to clients.
+     * Due to this, by default, Casssandra will prime these internode TCP connections and
wait for all but a single
+     * node to be DOWN/disconnected in the local datacenter before offering itself as a coordinator,
subject to a
+     * timeout. See CASSANDRA-13993 and CASSANDRA-14297 for more details.
+     *
+     * We provide two tunables to control this behavior as some users may want to block until
all datacenters are
+     * available (global QUORUM/EACH_QUORUM), some users may not want to block at all (clients
that already work
+     * around the problem), and some users may want to prime the connections but not delay
startup.
+     *
+     * block_for_peers_timeout_in_secs: controls how long this node will wait to connect
to peers. To completely disable
+     * any startup connectivity checks set this to -1. To trigger the internode connections
but immediately continue
+     * startup, set this to to 0. The default is 10 seconds.
+     *
+     * block_for_peers_in_remote_dcs: controls if this node will consider remote datacenters
to wait for. The default
+     * is to _not_ wait on remote datacenters.
+     */
     public int block_for_peers_timeout_in_secs = 10;
+    public boolean block_for_peers_in_remote_dcs = false;
+
     public volatile boolean automatic_sstable_upgrade = false;
     public volatile int max_concurrent_automatic_sstable_upgrades = 1;
     public boolean stream_entire_sstables = true;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/801cb70e/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 1b11a91..bc1e5a2 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -2630,9 +2630,9 @@ public class DatabaseDescriptor
         return  conf.full_query_logging_options;
     }
 
-    public static int getBlockForPeersPercentage()
+    public static boolean getBlockForPeersInRemoteDatacenters()
     {
-        return conf.block_for_peers_percentage;
+        return conf.block_for_peers_in_remote_dcs;
     }
 
     public static int getBlockForPeersTimeoutInSeconds()

http://git-wip-us.apache.org/repos/asf/cassandra/blob/801cb70e/src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java b/src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java
index bab3283..8e37470 100644
--- a/src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java
+++ b/src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java
@@ -17,6 +17,8 @@
  */
 package org.apache.cassandra.net;
 
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
@@ -24,9 +26,12 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Sets;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.SetMultimap;
 import com.google.common.util.concurrent.Uninterruptibles;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -48,73 +53,126 @@ public class StartupClusterConnectivityChecker
 {
     private static final Logger logger = LoggerFactory.getLogger(StartupClusterConnectivityChecker.class);
 
-    private final int targetPercent;
+    private final boolean blockForRemoteDcs;
     private final long timeoutNanos;
 
-    public static StartupClusterConnectivityChecker create(int targetPercent, int timeoutSecs)
+    public static StartupClusterConnectivityChecker create(long timeoutSecs, boolean blockForRemoteDcs)
     {
-        timeoutSecs = Math.max(1, timeoutSecs);
         if (timeoutSecs > 100)
             logger.warn("setting the block-for-peers timeout (in seconds) to {} might be
a bit excessive, but using it nonetheless", timeoutSecs);
         long timeoutNanos = TimeUnit.SECONDS.toNanos(timeoutSecs);
 
-        return new StartupClusterConnectivityChecker(targetPercent, timeoutNanos);
+        return new StartupClusterConnectivityChecker(timeoutNanos, blockForRemoteDcs);
     }
 
     @VisibleForTesting
-    StartupClusterConnectivityChecker(int targetPercent, long timeoutNanos)
+    StartupClusterConnectivityChecker(long timeoutNanos, boolean blockForRemoteDcs)
     {
-        this.targetPercent = Math.min(100, Math.max(0, targetPercent));
+        this.blockForRemoteDcs = blockForRemoteDcs;
         this.timeoutNanos = timeoutNanos;
     }
 
     /**
      * @param peers The currently known peers in the cluster; argument is not modified.
+     * @param getDatacenterSource A function for mapping peers to their datacenter.
      * @return true if the requested percentage of peers are marked ALIVE in gossip and have
their connections opened;
      * else false.
      */
-    public boolean execute(Set<InetAddressAndPort> peers)
+    public boolean execute(Set<InetAddressAndPort> peers, Function<InetAddressAndPort,
String> getDatacenterSource)
     {
-        if (targetPercent == 0 || peers == null)
+        if (peers == null || this.timeoutNanos < 0)
             return true;
 
         // make a copy of the set, to avoid mucking with the input (in case it's a sensitive
collection)
         peers = new HashSet<>(peers);
-        peers.remove(FBUtilities.getBroadcastAddressAndPort());
+        InetAddressAndPort localAddress = FBUtilities.getBroadcastAddressAndPort();
+        String localDc = getDatacenterSource.apply(localAddress);
 
+        peers.remove(localAddress);
         if (peers.isEmpty())
             return true;
 
-        logger.info("choosing to block until {}% of the {} known peers are marked alive and
connections are established; max time to wait = {} seconds",
-                    targetPercent, peers.size(), TimeUnit.NANOSECONDS.toSeconds(timeoutNanos));
+        // make a copy of the datacenter mapping (in case gossip updates happen during this
method or some such)
+        Map<InetAddressAndPort, String> peerToDatacenter = new HashMap<>();
+        SetMultimap<String, InetAddressAndPort> datacenterToPeers = HashMultimap.create();
 
-        long startNanos = System.nanoTime();
+        for (InetAddressAndPort peer : peers)
+        {
+            String datacenter = getDatacenterSource.apply(peer);
+            peerToDatacenter.put(peer, datacenter);
+            datacenterToPeers.put(datacenter, peer);
+        }
+
+        // In the case where we do not want to block startup on remote datacenters (e.g.
because clients only use
+        // LOCAL_X consistency levels), we remove all other datacenter hosts from the mapping
and we only wait
+        // on the remaining local datacenter.
+        if (!blockForRemoteDcs)
+        {
+            datacenterToPeers.keySet().retainAll(Collections.singleton(localDc));
+            logger.info("Blocking coordination until only a single peer is DOWN in the local
datacenter, timeout={}s",
+                        TimeUnit.NANOSECONDS.toSeconds(timeoutNanos));
+        }
+        else
+        {
+            logger.info("Blocking coordination until only a single peer is DOWN in each datacenter,
timeout={}s",
+                        TimeUnit.NANOSECONDS.toSeconds(timeoutNanos));
+        }
 
         AckMap acks = new AckMap(3);
-        int target = (int) ((targetPercent / 100.0) * peers.size());
-        CountDownLatch latch = new CountDownLatch(target);
+        Map<String, CountDownLatch> dcToRemainingPeers = new HashMap<>(datacenterToPeers.size());
+        for (String datacenter: datacenterToPeers.keys())
+        {
+            dcToRemainingPeers.put(datacenter,
+                                   new CountDownLatch(Math.max(datacenterToPeers.get(datacenter).size()
- 1, 0)));
+        }
+
+        long startNanos = System.nanoTime();
 
         // set up a listener to react to new nodes becoming alive (in gossip), and account
for all the nodes that are already alive
-        Set<InetAddressAndPort> alivePeers = Sets.newSetFromMap(new ConcurrentHashMap<>());
-        AliveListener listener = new AliveListener(alivePeers, latch, acks);
+        Set<InetAddressAndPort> alivePeers = Collections.newSetFromMap(new ConcurrentHashMap<>());
+        AliveListener listener = new AliveListener(alivePeers, dcToRemainingPeers, acks,
peerToDatacenter::get);
         Gossiper.instance.register(listener);
 
-        // send out a ping message to open up the non-gossip connections
-        sendPingMessages(peers, latch, acks);
+        // send out a ping message to open up the non-gossip connections to all peers. Note
that this sends the
+        // ping messages to _all_ peers, not just the ones we block for in dcToRemainingPeers.
+        sendPingMessages(peers, dcToRemainingPeers, acks, peerToDatacenter::get);
 
         for (InetAddressAndPort peer : peers)
+        {
             if (Gossiper.instance.isAlive(peer) && alivePeers.add(peer) &&
acks.incrementAndCheck(peer))
-                latch.countDown();
+            {
+                String datacenter = peerToDatacenter.get(peer);
+                // We have to check because we might only have the local DC in the map
+                if (dcToRemainingPeers.containsKey(datacenter))
+                    dcToRemainingPeers.get(datacenter).countDown();
+            }
+        }
+
+        boolean succeeded = true;
+        for (String datacenter: dcToRemainingPeers.keySet())
+        {
+            long remainingNanos = Math.max(1, timeoutNanos - (System.nanoTime() - startNanos));
+            succeeded &= Uninterruptibles.awaitUninterruptibly(dcToRemainingPeers.get(datacenter),
+                                                               remainingNanos, TimeUnit.NANOSECONDS);
+        }
 
-        boolean succeeded = Uninterruptibles.awaitUninterruptibly(latch, timeoutNanos, TimeUnit.NANOSECONDS);
         Gossiper.instance.unregister(listener);
 
-        int connected = peers.size() - (int) latch.getCount();
-        logger.info("After waiting/processing for {} milliseconds, {} out of {} peers ({}%)
have been marked alive and had connections established",
-                    TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos),
-                    connected,
-                    peers.size(),
-                    String.format("%.2f", (connected / (float)peers.size()) * 100));
+        Map<String, Long> numDown = dcToRemainingPeers.entrySet().stream()
+                                                      .collect(Collectors.toMap(Map.Entry::getKey,
+                                                                                e -> e.getValue().getCount()));
+
+        if (succeeded)
+        {
+            logger.info("Ensured sufficient healthy connections with {} after {} milliseconds",
+                        numDown.keySet(), TimeUnit.NANOSECONDS.toMillis(System.nanoTime()
- startNanos));
+        }
+        else
+        {
+            logger.warn("Timed out after {} milliseconds, was waiting for remaining peers
to connect: {}",
+                        TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos), numDown);
+        }
+
         return succeeded;
     }
 
@@ -122,7 +180,8 @@ public class StartupClusterConnectivityChecker
      * Sends a "connection warmup" message to each peer in the collection, on every {@link
ConnectionType}
      * used for internode messaging (that is not gossip).
      */
-    private void sendPingMessages(Set<InetAddressAndPort> peers, CountDownLatch latch,
AckMap acks)
+    private void sendPingMessages(Set<InetAddressAndPort> peers, Map<String, CountDownLatch>
dcToRemainingPeers,
+                                  AckMap acks, Function<InetAddressAndPort, String>
getDatacenter)
     {
         IAsyncCallback responseHandler = new IAsyncCallback()
         {
@@ -134,7 +193,12 @@ public class StartupClusterConnectivityChecker
             public void response(MessageIn msg)
             {
                 if (acks.incrementAndCheck(msg.from))
-                    latch.countDown();
+                {
+                    String datacenter = getDatacenter.apply(msg.from);
+                    // We have to check because we might only have the local DC in the map
+                    if (dcToRemainingPeers.containsKey(datacenter))
+                        dcToRemainingPeers.get(datacenter).countDown();
+                }
             }
         };
 
@@ -155,15 +219,18 @@ public class StartupClusterConnectivityChecker
      */
     private static final class AliveListener implements IEndpointStateChangeSubscriber
     {
-        private final CountDownLatch latch;
+        private final Map<String, CountDownLatch> dcToRemainingPeers;
         private final Set<InetAddressAndPort> livePeers;
+        private final Function<InetAddressAndPort, String> getDatacenter;
         private final AckMap acks;
 
-        AliveListener(Set<InetAddressAndPort> livePeers, CountDownLatch latch, AckMap
acks)
+        AliveListener(Set<InetAddressAndPort> livePeers, Map<String, CountDownLatch>
dcToRemainingPeers,
+                      AckMap acks, Function<InetAddressAndPort, String> getDatacenter)
         {
-            this.latch = latch;
             this.livePeers = livePeers;
+            this.dcToRemainingPeers = dcToRemainingPeers;
             this.acks = acks;
+            this.getDatacenter = getDatacenter;
         }
 
         public void onJoin(InetAddressAndPort endpoint, EndpointState epState)
@@ -181,7 +248,11 @@ public class StartupClusterConnectivityChecker
         public void onAlive(InetAddressAndPort endpoint, EndpointState state)
         {
             if (livePeers.add(endpoint) && acks.incrementAndCheck(endpoint))
-                latch.countDown();
+            {
+                String datacenter = getDatacenter.apply(endpoint);
+                if (dcToRemainingPeers.containsKey(datacenter))
+                    dcToRemainingPeers.get(datacenter).countDown();
+            }
         }
 
         public void onDead(InetAddressAndPort endpoint, EndpointState state)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/801cb70e/src/java/org/apache/cassandra/service/CassandraDaemon.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index 815e673..f0b2dc1 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -505,9 +505,9 @@ public class CassandraDaemon
      */
     public void start()
     {
-        StartupClusterConnectivityChecker connectivityChecker = StartupClusterConnectivityChecker.create(DatabaseDescriptor.getBlockForPeersPercentage(),
-                                                                                        
                DatabaseDescriptor.getBlockForPeersTimeoutInSeconds());
-        connectivityChecker.execute(Gossiper.instance.getEndpoints());
+        StartupClusterConnectivityChecker connectivityChecker = StartupClusterConnectivityChecker.create(DatabaseDescriptor.getBlockForPeersTimeoutInSeconds(),
+                                                                                        
                DatabaseDescriptor.getBlockForPeersInRemoteDatacenters());
+        connectivityChecker.execute(Gossiper.instance.getEndpoints(), DatabaseDescriptor.getEndpointSnitch()::getDatacenter);
 
         String nativeFlag = System.getProperty("cassandra.start_native_transport");
         if ((nativeFlag != null && Boolean.parseBoolean(nativeFlag)) || (nativeFlag
== null && DatabaseDescriptor.startNativeTransport()))

http://git-wip-us.apache.org/repos/asf/cassandra/blob/801cb70e/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java
b/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java
index 4eeb314..1645d77 100644
--- a/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java
+++ b/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java
@@ -36,13 +36,35 @@ import org.apache.cassandra.gms.EndpointState;
 import org.apache.cassandra.gms.Gossiper;
 import org.apache.cassandra.gms.HeartBeatState;
 import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.utils.FBUtilities;
 
 import static org.apache.cassandra.net.async.OutboundConnectionIdentifier.ConnectionType.SMALL_MESSAGE;
 
 public class StartupClusterConnectivityCheckerTest
 {
-    private StartupClusterConnectivityChecker connectivityChecker;
+    private StartupClusterConnectivityChecker localQuorumConnectivityChecker;
+    private StartupClusterConnectivityChecker globalQuorumConnectivityChecker;
+    private StartupClusterConnectivityChecker noopChecker;
+    private StartupClusterConnectivityChecker zeroWaitChecker;
+
+    private static final long TIMEOUT_NANOS = 100;
+    private static final int NUM_PER_DC = 6;
     private Set<InetAddressAndPort> peers;
+    private Set<InetAddressAndPort> peersA;
+    private Set<InetAddressAndPort> peersAMinusLocal;
+    private Set<InetAddressAndPort> peersB;
+    private Set<InetAddressAndPort> peersC;
+
+    private String getDatacenter(InetAddressAndPort endpoint)
+    {
+        if (peersA.contains(endpoint))
+            return "datacenterA";
+        if (peersB.contains(endpoint))
+            return "datacenterB";
+        else if (peersC.contains(endpoint))
+            return "datacenterC";
+        return null;
+    }
 
     @BeforeClass
     public static void before()
@@ -53,11 +75,34 @@ public class StartupClusterConnectivityCheckerTest
     @Before
     public void setUp() throws UnknownHostException
     {
-        connectivityChecker = new StartupClusterConnectivityChecker(70, 10);
+        localQuorumConnectivityChecker = new StartupClusterConnectivityChecker(TIMEOUT_NANOS,
false);
+        globalQuorumConnectivityChecker = new StartupClusterConnectivityChecker(TIMEOUT_NANOS,
true);
+        noopChecker = new StartupClusterConnectivityChecker(-1, false);
+        zeroWaitChecker = new StartupClusterConnectivityChecker(0, false);
+
+        peersA = new HashSet<>();
+        peersAMinusLocal = new HashSet<>();
+        peersA.add(FBUtilities.getBroadcastAddressAndPort());
+
+        for (int i = 0; i < NUM_PER_DC - 1; i ++)
+        {
+            peersA.add(InetAddressAndPort.getByName("127.0.1." + i));
+            peersAMinusLocal.add(InetAddressAndPort.getByName("127.0.1." + i));
+        }
+
+        peersB = new HashSet<>();
+        for (int i = 0; i < NUM_PER_DC; i ++)
+            peersB.add(InetAddressAndPort.getByName("127.0.2." + i));
+
+
+        peersC = new HashSet<>();
+        for (int i = 0; i < NUM_PER_DC; i ++)
+            peersC.add(InetAddressAndPort.getByName("127.0.3." + i));
+
         peers = new HashSet<>();
-        peers.add(InetAddressAndPort.getByName("127.0.1.0"));
-        peers.add(InetAddressAndPort.getByName("127.0.1.1"));
-        peers.add(InetAddressAndPort.getByName("127.0.1.2"));
+        peers.addAll(peersA);
+        peers.addAll(peersB);
+        peers.addAll(peersC);
     }
 
     @After
@@ -69,50 +114,145 @@ public class StartupClusterConnectivityCheckerTest
     @Test
     public void execute_HappyPath()
     {
-        Sink sink = new Sink(true, true);
+        Sink sink = new Sink(true, true, peers);
         MessagingService.instance().addMessageSink(sink);
-        Assert.assertTrue(connectivityChecker.execute(peers));
-        checkAllConnectionTypesSeen(sink);
+        Assert.assertTrue(localQuorumConnectivityChecker.execute(peers, this::getDatacenter));
+        Assert.assertTrue(checkAllConnectionTypesSeen(sink));
     }
 
     @Test
     public void execute_NotAlive()
     {
-        Sink sink = new Sink(false, true);
+        Sink sink = new Sink(false, true, peers);
         MessagingService.instance().addMessageSink(sink);
-        Assert.assertFalse(connectivityChecker.execute(peers));
-        checkAllConnectionTypesSeen(sink);
+        Assert.assertFalse(localQuorumConnectivityChecker.execute(peers, this::getDatacenter));
+        Assert.assertTrue(checkAllConnectionTypesSeen(sink));
     }
 
     @Test
     public void execute_NoConnectionsAcks()
     {
-        Sink sink = new Sink(true, false);
+        Sink sink = new Sink(true, false, peers);
+        MessagingService.instance().addMessageSink(sink);
+        Assert.assertFalse(localQuorumConnectivityChecker.execute(peers, this::getDatacenter));
+    }
+
+    @Test
+    public void execute_LocalQuorum()
+    {
+        // local peer plus 3 peers from same dc shouldn't pass (4/6)
+        Set<InetAddressAndPort> available = new HashSet<>();
+        copyCount(peersAMinusLocal, available, NUM_PER_DC - 3);
+        checkAvailable(localQuorumConnectivityChecker, available, false, true);
+
+        // local peer plus 4 peers from same dc should pass (5/6)
+        available.clear();
+        copyCount(peersAMinusLocal, available, NUM_PER_DC - 2);
+        checkAvailable(localQuorumConnectivityChecker, available, true, true);
+    }
+
+    @Test
+    public void execute_GlobalQuorum()
+    {
+        // local dc passing shouldn't pass globally with two hosts down in datacenterB
+        Set<InetAddressAndPort> available = new HashSet<>();
+        copyCount(peersAMinusLocal, available, NUM_PER_DC - 2);
+        copyCount(peersB, available, NUM_PER_DC - 2);
+        copyCount(peersC, available, NUM_PER_DC - 1);
+        checkAvailable(globalQuorumConnectivityChecker, available, false, true);
+
+        // All three datacenters should be able to have a single node down
+        available.clear();
+        copyCount(peersAMinusLocal, available, NUM_PER_DC - 2);
+        copyCount(peersB, available, NUM_PER_DC - 1);
+        copyCount(peersC, available, NUM_PER_DC - 1);
+        checkAvailable(globalQuorumConnectivityChecker, available, true, true);
+
+        // Everything being up should work of course
+        available.clear();
+        copyCount(peersAMinusLocal, available, NUM_PER_DC - 1);
+        copyCount(peersB, available, NUM_PER_DC);
+        copyCount(peersC, available, NUM_PER_DC);
+        checkAvailable(globalQuorumConnectivityChecker, available, true, true);
+    }
+
+    @Test
+    public void execute_Noop()
+    {
+        checkAvailable(noopChecker, new HashSet<>(), true, false);
+    }
+
+    @Test
+    public void execute_ZeroWaitHasConnections() throws InterruptedException
+    {
+        Sink sink = new Sink(true, true, new HashSet<>());
+        MessagingService.instance().addMessageSink(sink);
+        Assert.assertFalse(zeroWaitChecker.execute(peers, this::getDatacenter));
+        boolean hasConnections = false;
+        for (int i = 0; i < TIMEOUT_NANOS; i+= 10)
+        {
+            hasConnections = checkAllConnectionTypesSeen(sink);
+            if (hasConnections)
+                break;
+            Thread.sleep(0, 10);
+        }
+        MessagingService.instance().clearMessageSinks();
+        Assert.assertTrue(hasConnections);
+    }
+
+    private void checkAvailable(StartupClusterConnectivityChecker checker, Set<InetAddressAndPort>
available,
+                                boolean shouldPass, boolean checkConnections)
+    {
+        Sink sink = new Sink(true, true, available);
         MessagingService.instance().addMessageSink(sink);
-        Assert.assertFalse(connectivityChecker.execute(peers));
+        Assert.assertEquals(shouldPass, checker.execute(peers, this::getDatacenter));
+        if (checkConnections)
+            Assert.assertTrue(checkAllConnectionTypesSeen(sink));
+        MessagingService.instance().clearMessageSinks();
+    }
+
+    private void copyCount(Set<InetAddressAndPort> source, Set<InetAddressAndPort>
dest, int count)
+    {
+        for (InetAddressAndPort peer : source)
+        {
+            if (count <= 0)
+                break;
+
+            dest.add(peer);
+            count -= 1;
+        }
     }
 
-    private void checkAllConnectionTypesSeen(Sink sink)
+    private boolean checkAllConnectionTypesSeen(Sink sink)
     {
+        boolean result = true;
         for (InetAddressAndPort peer : peers)
         {
+            if (peer.equals(FBUtilities.getBroadcastAddressAndPort()))
+                continue;
             ConnectionTypeRecorder recorder = sink.seenConnectionRequests.get(peer);
-            Assert.assertNotNull(recorder);
-            Assert.assertTrue(recorder.seenSmallMessageRequest);
-            Assert.assertTrue(recorder.seenLargeMessageRequest);
+            result = recorder != null;
+            if (!result)
+                break;
+
+            result = recorder.seenSmallMessageRequest;
+            result &= recorder.seenLargeMessageRequest;
         }
+        return result;
     }
 
     private static class Sink implements IMessageSink
     {
         private final boolean markAliveInGossip;
         private final boolean processConnectAck;
+        private final Set<InetAddressAndPort> aliveHosts;
         private final Map<InetAddressAndPort, ConnectionTypeRecorder> seenConnectionRequests;
 
-        Sink(boolean markAliveInGossip, boolean processConnectAck)
+        Sink(boolean markAliveInGossip, boolean processConnectAck, Set<InetAddressAndPort>
aliveHosts)
         {
             this.markAliveInGossip = markAliveInGossip;
             this.processConnectAck = processConnectAck;
+            this.aliveHosts = aliveHosts;
             seenConnectionRequests = new HashMap<>();
         }
 
@@ -131,6 +271,9 @@ public class StartupClusterConnectivityCheckerTest
                 recorder.seenLargeMessageRequest = true;
             }
 
+            if (!aliveHosts.contains(to))
+                return false;
+
             if (processConnectAck)
             {
                 MessageIn msgIn = MessageIn.create(to, message.payload, Collections.emptyMap(),
MessagingService.Verb.REQUEST_RESPONSE, 1);


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message