natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make StartupConnectivityChecker only run a connectivity check if there are no nodes which are running a version prior to Cassandra 4 

patch by Isaac Reath; reviwed by Paulo Motta, Stefan Miklosovic for CASSANDRA-18968
This commit is contained in:
Isaac Reath 2023-11-10 11:08:32 +01:00 committed by Stefan Miklosovic
parent ac8a8687a3
commit 26a7d574f0
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
4 changed files with 52 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0.12
* Make StartupConnectivityChecker only run a connectivity check if there are no nodes which are running a version prior to Cassandra 4 (CASSANDRA-18968)
* Retrieve keyspaces metadata and schema version concistently in DescribeStatement (CASSANDRA-18921)
* Gossip NPE due to shutdown event corrupting empty statuses (CASSANDRA-18913)
* Synchronize CQLSSTableWriter#build on the Schema.instance object (CASSANDRA-18317)

17
src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java
View File

@ -27,6 +27,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
@ -36,12 +37,11 @@ import com.google.common.util.concurrent.Uninterruptibles;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.IEndpointStateChangeSubscriber;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.utils.CassandraVersion;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.net.Verb.PING_REQ;
@ -77,11 +77,22 @@ public class StartupClusterConnectivityChecker
* @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, Function<InetAddressAndPort, String> getDatacenterSource)
public boolean execute(Set<InetAddressAndPort> peers, Function<InetAddressAndPort, String> getDatacenterSource,
Predicate<CassandraVersion> isUpgradingFromLowerVersionThan)
{
if (peers == null || this.timeoutNanos < 0)
return true;
// Check if there are any nodes which we know are running a version prior to 4.0.
// We use this intead of Gossiper::hasMajorVersion3Nodes because in the absence of version information for a peer
// we still prefer to run the startup connectivity check.
if (isUpgradingFromLowerVersionThan.test(CassandraVersion.CASSANDRA_4_0))
{
logger.debug("Skipping startup connectivity check as some nodes may be running Cassandra version 3 or older " +
"which does not support connectivity checking.");
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);
InetAddressAndPort localAddress = FBUtilities.getBroadcastAddressAndPort();

3
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -667,7 +667,8 @@ public class CassandraDaemon
{
StartupClusterConnectivityChecker connectivityChecker = StartupClusterConnectivityChecker.create(DatabaseDescriptor.getBlockForPeersTimeoutInSeconds(),
DatabaseDescriptor.getBlockForPeersInRemoteDatacenters());
connectivityChecker.execute(Gossiper.instance.getEndpoints(), DatabaseDescriptor.getEndpointSnitch()::getDatacenter);
connectivityChecker.execute(Gossiper.instance.getEndpoints(), DatabaseDescriptor.getEndpointSnitch()::getDatacenter,
Gossiper.instance::isUpgradingFromVersionLowerThan);
// check to see if transports may start else return without starting. This is needed when in survey mode or
// when bootstrap has not completed.

41
test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java
View File

@ -24,6 +24,7 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.function.BiPredicate;
import java.util.function.Predicate;
import org.junit.After;
import org.junit.Assert;
@ -36,6 +37,7 @@ 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.CassandraVersion;
import org.apache.cassandra.utils.FBUtilities;
public class StartupClusterConnectivityCheckerTest
@ -44,6 +46,8 @@ public class StartupClusterConnectivityCheckerTest
private StartupClusterConnectivityChecker globalQuorumConnectivityChecker;
private StartupClusterConnectivityChecker noopChecker;
private StartupClusterConnectivityChecker zeroWaitChecker;
private StartupClusterConnectivityChecker containsCassandra3NodesChecker;
private Predicate<CassandraVersion> defaultMockIsUpgradingFromVersionLowerThan;
private static final long TIMEOUT_NANOS = 100;
private static final int NUM_PER_DC = 6;
@ -64,6 +68,20 @@ public class StartupClusterConnectivityCheckerTest
return null;
}
private static class MockIsUpgradingFromVersionLowerThan implements Predicate<CassandraVersion>
{
CassandraVersion clusterVersion;
MockIsUpgradingFromVersionLowerThan(CassandraVersion clusterVersion)
{
this.clusterVersion = clusterVersion;
}
@Override
public boolean test(CassandraVersion other)
{
return clusterVersion.compareTo(other) < 0;
}
}
@BeforeClass
public static void before()
{
@ -77,7 +95,8 @@ public class StartupClusterConnectivityCheckerTest
globalQuorumConnectivityChecker = new StartupClusterConnectivityChecker(TIMEOUT_NANOS, true);
noopChecker = new StartupClusterConnectivityChecker(-1, false);
zeroWaitChecker = new StartupClusterConnectivityChecker(0, false);
containsCassandra3NodesChecker = new StartupClusterConnectivityChecker(TIMEOUT_NANOS, false);
defaultMockIsUpgradingFromVersionLowerThan = new MockIsUpgradingFromVersionLowerThan(CassandraVersion.CASSANDRA_4_0);
peersA = new HashSet<>();
peersAMinusLocal = new HashSet<>();
peersA.add(FBUtilities.getBroadcastAddressAndPort());
@ -114,7 +133,7 @@ public class StartupClusterConnectivityCheckerTest
{
Sink sink = new Sink(true, true, peers);
MessagingService.instance().outboundSink.add(sink);
Assert.assertTrue(localQuorumConnectivityChecker.execute(peers, this::getDatacenter));
Assert.assertTrue(localQuorumConnectivityChecker.execute(peers, this::getDatacenter, defaultMockIsUpgradingFromVersionLowerThan));
}
@Test
@ -122,7 +141,7 @@ public class StartupClusterConnectivityCheckerTest
{
Sink sink = new Sink(false, true, peers);
MessagingService.instance().outboundSink.add(sink);
Assert.assertFalse(localQuorumConnectivityChecker.execute(peers, this::getDatacenter));
Assert.assertFalse(localQuorumConnectivityChecker.execute(peers, this::getDatacenter, defaultMockIsUpgradingFromVersionLowerThan));
}
@Test
@ -130,7 +149,7 @@ public class StartupClusterConnectivityCheckerTest
{
Sink sink = new Sink(true, false, peers);
MessagingService.instance().outboundSink.add(sink);
Assert.assertFalse(localQuorumConnectivityChecker.execute(peers, this::getDatacenter));
Assert.assertFalse(localQuorumConnectivityChecker.execute(peers, this::getDatacenter, defaultMockIsUpgradingFromVersionLowerThan));
}
@Test
@ -183,16 +202,26 @@ public class StartupClusterConnectivityCheckerTest
{
Sink sink = new Sink(true, true, new HashSet<>());
MessagingService.instance().outboundSink.add(sink);
Assert.assertFalse(zeroWaitChecker.execute(peers, this::getDatacenter));
Assert.assertFalse(zeroWaitChecker.execute(peers, this::getDatacenter, defaultMockIsUpgradingFromVersionLowerThan));
MessagingService.instance().outboundSink.clear();
}
@Test
public void execute_HasCassandra3NodesSkipsExecution()
{
Sink sink = new Sink(true, true, peers);
MessagingService.instance().outboundSink.add(sink);
Predicate<CassandraVersion> isUpgradingFromVersionLowerThan = new MockIsUpgradingFromVersionLowerThan(new CassandraVersion("3.11.0"));
Assert.assertTrue(containsCassandra3NodesChecker.execute(peers, this::getDatacenter, isUpgradingFromVersionLowerThan));
Assert.assertEquals(0, sink.seenConnectionRequests.size());
}
private void checkAvailable(StartupClusterConnectivityChecker checker, Set<InetAddressAndPort> available,
boolean shouldPass)
{
Sink sink = new Sink(true, true, available);
MessagingService.instance().outboundSink.add(sink);
Assert.assertEquals(shouldPass, checker.execute(peers, this::getDatacenter));
Assert.assertEquals(shouldPass, checker.execute(peers, this::getDatacenter, defaultMockIsUpgradingFromVersionLowerThan));
MessagingService.instance().outboundSink.clear();
}