This is an automated email from the ASF dual-hosted git repository. bdeggleston pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 587961a19c493efe6b9b065d47416aa4b62cc60a Merge: 79766f7 01d6548 Author: Blake Eggleston <[email protected]> AuthorDate: Thu Apr 25 10:23:11 2019 -0700 Merge branch 'cassandra-3.11' into trunk CHANGES.txt | 1 + build.xml | 1 + ide/idea/workspace.xml | 2 +- .../cassandra/config/DatabaseDescriptor.java | 8 + src/java/org/apache/cassandra/gms/Gossiper.java | 201 ++++++++++++++------- .../apache/cassandra/service/StorageService.java | 10 +- .../cassandra/distributed/impl/Instance.java | 26 +-- .../org/apache/cassandra/gms/GossiperTest.java | 1 + .../gms/PendingRangeCalculatorServiceTest.java | 1 + .../cassandra/locator/CloudstackSnitchTest.java | 1 + .../apache/cassandra/locator/EC2SnitchTest.java | 1 + .../cassandra/locator/GoogleCloudSnitchTest.java | 1 + .../cassandra/locator/PropertyFileSnitchTest.java | 1 + .../net/StartupClusterConnectivityCheckerTest.java | 2 +- .../service/StorageServiceServerTest.java | 1 + 15 files changed, 179 insertions(+), 79 deletions(-) diff --cc CHANGES.txt index 28eb042,0506da7..54ffe5b --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -360,6 -2,8 +360,7 @@@ * Fixed nodetool cfstats printing index name twice (CASSANDRA-14903) * Add flag to disable SASI indexes, and warnings on creation (CASSANDRA-14866) Merged from 3.0: + * Fix assorted gossip races and add related runtime checks (CASSANDRA-15059) - * Fix mixed mode partition range scans with limit (CASSANDRA-15072) * cassandra-stress works with frozen collections: list and set (CASSANDRA-14907) * Fix handling FS errors on writing and reading flat files - LogTransaction and hints (CASSANDRA-15053) * Avoid double closing the iterator to avoid overcounting the number of requests (CASSANDRA-15058) diff --cc build.xml index 3a8dd6c,db97d05..4b23a25 --- a/build.xml +++ b/build.xml @@@ -1334,12 -1274,8 +1334,13 @@@ <jvmarg value="-Djava.security.egd=file:/dev/urandom" /> <jvmarg value="-Dcassandra.testtag=@{testtag}"/> <jvmarg value="-Dcassandra.keepBriefBrief=${cassandra.keepBriefBrief}" /> - <jvmarg value="-Dcassandra.strict.runtime.checks=true" /> - <optjvmargs/> ++ <jvmarg value="-Dcassandra.strict.runtime.checks=true" /> + <jvmarg line="${java11-jvmargs}"/> + <optjvmargs/> + <!-- Uncomment to debug unittest, attach debugger to port 1416 --> + <!-- + <jvmarg line="-agentlib:jdwp=transport=dt_socket,address=localhost:1416,server=y,suspend=y" /> + --> <classpath> <pathelement path="${java.class.path}"/> <pathelement location="${stress.build.classes}"/> diff --cc ide/idea/workspace.xml index 150f1a0,8d1b0fc..378997b --- a/ide/idea/workspace.xml +++ b/ide/idea/workspace.xml @@@ -167,7 -167,7 +167,7 @@@ <option name="MAIN_CLASS_NAME" value="" /> <option name="METHOD_NAME" value="" /> <option name="TEST_OBJECT" value="class" /> - <option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.skip_sync=true -ea -XX:MaxMe [...] - <option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -ea -XX:MaxMetaspaceSize=256M -XX:SoftRefLRUPolicyMSPerMB=0 -Dcassandra.strict.runtime.checks=true" /> ++ <option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.skip_sync=true -ea -XX:MaxMe [...] <option name="PARAMETERS" value="" /> <option name="WORKING_DIRECTORY" value="" /> <option name="ENV_VARIABLES" /> diff --cc src/java/org/apache/cassandra/config/DatabaseDescriptor.java index e5fe772,e452830..e2c2ace --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@@ -2690,121 -2533,8 +2693,126 @@@ public class DatabaseDescripto return backPressureStrategy; } + public static ConsistencyLevel getIdealConsistencyLevel() + { + return conf.ideal_consistency_level; + } + + public static void setIdealConsistencyLevel(ConsistencyLevel cl) + { + conf.ideal_consistency_level = cl; + } + + public static int getRepairCommandPoolSize() + { + return conf.repair_command_pool_size; + } + + public static Config.RepairCommandPoolFullStrategy getRepairCommandPoolFullStrategy() + { + return conf.repair_command_pool_full_strategy; + } + + public static FullQueryLoggerOptions getFullQueryLogOptions() + { + return conf.full_query_logging_options; + } + + public static boolean getBlockForPeersInRemoteDatacenters() + { + return conf.block_for_peers_in_remote_dcs; + } + + public static int getBlockForPeersTimeoutInSeconds() + { + return conf.block_for_peers_timeout_in_secs; + } + + public static boolean automaticSSTableUpgrade() + { + return conf.automatic_sstable_upgrade; + } + + public static void setAutomaticSSTableUpgradeEnabled(boolean enabled) + { + if (conf.automatic_sstable_upgrade != enabled) + logger.debug("Changing automatic_sstable_upgrade to {}", enabled); + conf.automatic_sstable_upgrade = enabled; + } + + public static int maxConcurrentAutoUpgradeTasks() + { + return conf.max_concurrent_automatic_sstable_upgrades; + } + + public static void setMaxConcurrentAutoUpgradeTasks(int value) + { + if (conf.max_concurrent_automatic_sstable_upgrades != value) + logger.debug("Changing max_concurrent_automatic_sstable_upgrades to {}", value); + validateMaxConcurrentAutoUpgradeTasksConf(value); + conf.max_concurrent_automatic_sstable_upgrades = value; + } + + private static void validateMaxConcurrentAutoUpgradeTasksConf(int value) + { + if (value < 0) + throw new ConfigurationException("max_concurrent_automatic_sstable_upgrades can't be negative"); + if (value > getConcurrentCompactors()) + logger.warn("max_concurrent_automatic_sstable_upgrades ({}) is larger than concurrent_compactors ({})", value, getConcurrentCompactors()); + } + + public static AuditLogOptions getAuditLoggingOptions() + { + return conf.audit_logging_options; + } + + public static void setAuditLoggingOptions(AuditLogOptions auditLoggingOptions) + { + conf.audit_logging_options = auditLoggingOptions; + } + + public static Config.CorruptedTombstoneStrategy getCorruptedTombstoneStrategy() + { + return conf.corrupted_tombstone_strategy; + } + + public static void setCorruptedTombstoneStrategy(Config.CorruptedTombstoneStrategy strategy) + { + conf.corrupted_tombstone_strategy = strategy; + } + + public static boolean getRepairedDataTrackingForRangeReadsEnabled() + { + return conf.repaired_data_tracking_for_range_reads_enabled; + } + + public static void setRepairedDataTrackingForRangeReadsEnabled(boolean enabled) + { + conf.repaired_data_tracking_for_range_reads_enabled = enabled; + } + + public static boolean getRepairedDataTrackingForPartitionReadsEnabled() + { + return conf.repaired_data_tracking_for_partition_reads_enabled; + } + + public static void setRepairedDataTrackingForPartitionReadsEnabled(boolean enabled) + { + conf.repaired_data_tracking_for_partition_reads_enabled = enabled; + } + + public static boolean reportUnconfirmedRepairedDataMismatches() + { + return conf.report_unconfirmed_repaired_data_mismatches; + } + + public static void reportUnconfirmedRepairedDataMismatches(boolean enabled) + { + conf.report_unconfirmed_repaired_data_mismatches = enabled; + } ++ + public static boolean strictRuntimeChecks() + { + return strictRuntimeChecks; + } } diff --cc src/java/org/apache/cassandra/gms/Gossiper.java index b789fe7,5d2e997..8955bf9 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@@ -29,16 -27,15 +29,20 @@@ import java.util.stream.Collectors import javax.annotation.Nullable; import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Suppliers; + import com.google.common.base.Throwables; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Iterables; +import com.google.common.collect.ImmutableSet; + import com.google.common.util.concurrent.ListenableFutureTask; import com.google.common.util.concurrent.Uninterruptibles; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.utils.CassandraVersion; + import io.netty.util.concurrent.FastThreadLocal; import org.apache.cassandra.utils.MBeanWrapper; + import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@@ -92,7 -98,8 +102,9 @@@ public class Gossiper implements IFailu public final static int intervalInMillis = 1000; public final static int QUARANTINE_DELAY = StorageService.RING_DELAY * 2; private static final Logger logger = LoggerFactory.getLogger(Gossiper.class); + private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 15L, TimeUnit.MINUTES); - public static final Gossiper instance = new Gossiper(); ++ + public static final Gossiper instance = new Gossiper(true); // Timestamp to prevent processing any in-flight messages for we've not send any SYN yet, see CASSANDRA-12653. volatile long firstSynSendAt = 0L; @@@ -137,39 -151,37 +149,70 @@@ private volatile long lastProcessedMessageAt = System.currentTimeMillis(); + //This property and anything that checks it should be removed in 5.0 + private boolean haveMajorVersion3Nodes = true; + + final com.google.common.base.Supplier<Boolean> haveMajorVersion3NodesSupplier = () -> + { + //Once there are no prior version nodes we don't need to keep rechecking + if (!haveMajorVersion3Nodes) + return false; + + Iterable<InetAddressAndPort> allHosts = Iterables.concat(Gossiper.instance.getLiveMembers(), Gossiper.instance.getUnreachableMembers()); + CassandraVersion referenceVersion = null; + + for (InetAddressAndPort host : allHosts) + { + CassandraVersion version = getReleaseVersion(host); + + //Raced with changes to gossip state + if (version == null) + continue; + + if (referenceVersion == null) + referenceVersion = version; + + if (version.major < 4) + return true; + } + + haveMajorVersion3Nodes = false; + return false; + }; + + private final Supplier<Boolean> haveMajorVersion3NodesMemoized = Suppliers.memoizeWithExpiration(haveMajorVersion3NodesSupplier, 1, TimeUnit.MINUTES); + + private static FastThreadLocal<Boolean> isGossipStage = new FastThreadLocal<>(); + + private static final boolean disableThreadValidation = Boolean.getBoolean(Props.DISABLE_THREAD_VALIDATION); + + private static boolean isInGossipStage() + { + Boolean isGossip = isGossipStage.get(); + if (isGossip == null) + { + isGossip = Thread.currentThread().getName().contains(Stage.GOSSIP.getJmxName()); + isGossipStage.set(isGossip); + } + return isGossip; + } + + private static void checkProperThreadForStateMutation() + { + if (disableThreadValidation || isInGossipStage()) + return; + + IllegalStateException e = new IllegalStateException("Attempting gossip state mutation from illegal thread: " + Thread.currentThread().getName()); + if (DatabaseDescriptor.strictRuntimeChecks()) + { + throw e; + } + else + { + noSpamLogger.getStatement(Throwables.getStackTraceAsString(e)).error(e.getMessage(), e); + } + } + private class GossipTask implements Runnable { public void run() @@@ -387,36 -396,37 +451,37 @@@ * * @param endpoint end point that is convicted. */ - public void convict(InetAddress endpoint, double phi) + public void convict(InetAddressAndPort endpoint, double phi) { - EndpointState epState = endpointStateMap.get(endpoint); - if (epState == null) - return; - - if (!epState.isAlive()) - return; - - logger.debug("Convicting {} with status {} - alive {}", endpoint, getGossipStatus(epState), epState.isAlive()); + runInGossipStageBlocking(() -> { + EndpointState epState = endpointStateMap.get(endpoint); + if (epState == null) + return; + if (!epState.isAlive()) + return; - if (isShutdown(endpoint)) - { - markAsShutdown(endpoint); - } - else - { - markDead(endpoint, epState); - } + logger.debug("Convicting {} with status {} - alive {}", endpoint, getGossipStatus(epState), epState.isAlive()); - GossiperDiagnostics.convicted(this, endpoint, phi); - + if (isShutdown(endpoint)) + { + markAsShutdown(endpoint); + } + else + { + markDead(endpoint, epState); + } ++ GossiperDiagnostics.convicted(this, endpoint, phi); + }); } /** * This method is used to mark a node as shutdown; that is it gracefully exited on its own and told us about it * @param endpoint endpoint that has shut itself down */ - protected void markAsShutdown(InetAddress endpoint) + protected void markAsShutdown(InetAddressAndPort endpoint) { + checkProperThreadForStateMutation(); EndpointState epState = endpointStateMap.get(endpoint); if (epState == null) return; @@@ -448,8 -456,9 +513,9 @@@ * * @param endpoint endpoint to be removed from the current membership. */ - private void evictFromMembership(InetAddress endpoint) + private void evictFromMembership(InetAddressAndPort endpoint) { + checkProperThreadForStateMutation(); unreachableEndpoints.remove(endpoint); endpointStateMap.remove(endpoint); expireTimeEndpointMap.remove(endpoint); @@@ -463,8 -471,9 +529,9 @@@ /** * Removes the endpoint from Gossip but retains endpoint state */ - public void removeEndpoint(InetAddress endpoint) + public void removeEndpoint(InetAddressAndPort endpoint) { + checkProperThreadForStateMutation(); // do subscribers first so anything in the subscriber that depends on gossiper state won't get confused for (IEndpointStateChangeSubscriber subscriber : subscribers) subscriber.onRemove(endpoint); @@@ -526,8 -532,9 +593,9 @@@ * * @param endpoint The endpoint that has been replaced */ - public void replacedEndpoint(InetAddress endpoint) + public void replacedEndpoint(InetAddressAndPort endpoint) { + checkProperThreadForStateMutation(); removeEndpoint(endpoint); evictFromMembership(endpoint); replacementQuarantine(endpoint); @@@ -642,55 -646,55 +710,57 @@@ */ public void assassinateEndpoint(String address) throws UnknownHostException { - InetAddress endpoint = InetAddress.getByName(address); + InetAddressAndPort endpoint = InetAddressAndPort.getByName(address); - EndpointState epState = endpointStateMap.get(endpoint); - Collection<Token> tokens = null; - logger.warn("Assassinating {} via gossip", endpoint); + runInGossipStageBlocking(() -> { + EndpointState epState = endpointStateMap.get(endpoint); + Collection<Token> tokens = null; + logger.warn("Assassinating {} via gossip", endpoint); - if (epState == null) - { - epState = new EndpointState(new HeartBeatState((int) ((System.currentTimeMillis() + 60000) / 1000), 9999)); - } - else - { - int generation = epState.getHeartBeatState().getGeneration(); - int heartbeat = epState.getHeartBeatState().getHeartBeatVersion(); - logger.info("Sleeping for {}ms to ensure {} does not change", StorageService.RING_DELAY, endpoint); - Uninterruptibles.sleepUninterruptibly(StorageService.RING_DELAY, TimeUnit.MILLISECONDS); - // make sure it did not change - EndpointState newState = endpointStateMap.get(endpoint); - if (newState == null) - logger.warn("Endpoint {} disappeared while trying to assassinate, continuing anyway", endpoint); - else if (newState.getHeartBeatState().getGeneration() != generation) - throw new RuntimeException("Endpoint still alive: " + endpoint + " generation changed while trying to assassinate it"); - else if (newState.getHeartBeatState().getHeartBeatVersion() != heartbeat) - throw new RuntimeException("Endpoint still alive: " + endpoint + " heartbeat changed while trying to assassinate it"); - epState.updateTimestamp(); // make sure we don't evict it too soon - epState.getHeartBeatState().forceNewerGenerationUnsafe(); - } + if (epState == null) + { + epState = new EndpointState(new HeartBeatState((int) ((System.currentTimeMillis() + 60000) / 1000), 9999)); + } + else + { + int generation = epState.getHeartBeatState().getGeneration(); + int heartbeat = epState.getHeartBeatState().getHeartBeatVersion(); + logger.info("Sleeping for {}ms to ensure {} does not change", StorageService.RING_DELAY, endpoint); + Uninterruptibles.sleepUninterruptibly(StorageService.RING_DELAY, TimeUnit.MILLISECONDS); + // make sure it did not change + EndpointState newState = endpointStateMap.get(endpoint); + if (newState == null) + logger.warn("Endpoint {} disappeared while trying to assassinate, continuing anyway", endpoint); + else if (newState.getHeartBeatState().getGeneration() != generation) + throw new RuntimeException("Endpoint still alive: " + endpoint + " generation changed while trying to assassinate it"); + else if (newState.getHeartBeatState().getHeartBeatVersion() != heartbeat) + throw new RuntimeException("Endpoint still alive: " + endpoint + " heartbeat changed while trying to assassinate it"); + epState.updateTimestamp(); // make sure we don't evict it too soon + epState.getHeartBeatState().forceNewerGenerationUnsafe(); + } - try - { - tokens = StorageService.instance.getTokenMetadata().getTokens(endpoint); - } - catch (Throwable th) - { - JVMStabilityInspector.inspectThrowable(th); - // TODO this is broken - logger.warn("Unable to calculate tokens for {}. Will use a random one", address); - tokens = Collections.singletonList(StorageService.instance.getTokenMetadata().partitioner.getRandomToken()); - } + try + { + tokens = StorageService.instance.getTokenMetadata().getTokens(endpoint); + } + catch (Throwable th) + { + JVMStabilityInspector.inspectThrowable(th); + // TODO this is broken + logger.warn("Unable to calculate tokens for {}. Will use a random one", address); + tokens = Collections.singletonList(StorageService.instance.getTokenMetadata().partitioner.getRandomToken()); + } - // do not pass go, do not collect 200 dollars, just gtfo - long expireTime = computeExpireTime(); - epState.addApplicationState(ApplicationState.STATUS_WITH_PORT, StorageService.instance.valueFactory.left(tokens, expireTime)); - epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.left(tokens, expireTime)); - handleMajorStateChange(endpoint, epState); - Uninterruptibles.sleepUninterruptibly(intervalInMillis * 4, TimeUnit.MILLISECONDS); - logger.warn("Finished assassinating {}", endpoint); + // do not pass go, do not collect 200 dollars, just gtfo ++ long expireTime = computeExpireTime(); ++ epState.addApplicationState(ApplicationState.STATUS_WITH_PORT, StorageService.instance.valueFactory.left(tokens, expireTime)); + epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.left(tokens, computeExpireTime())); + handleMajorStateChange(endpoint, epState); + Uninterruptibles.sleepUninterruptibly(intervalInMillis * 4, TimeUnit.MILLISECONDS); + logger.warn("Finished assassinating {}", endpoint); + }); } - public boolean isKnownEndpoint(InetAddress endpoint) + public boolean isKnownEndpoint(InetAddressAndPort endpoint) { return endpointStateMap.containsKey(endpoint); } @@@ -1084,8 -1063,7 +1156,8 @@@ public void response(MessageIn msg) { + // force processing of the echo response onto the gossip stage, as it comes in on the REQUEST_RESPONSE stage - StageManager.getStage(Stage.GOSSIP).submit(() -> realMarkAlive(addr, localState)); + runInGossipStageBlocking(() -> realMarkAlive(addr, localState)); } }; @@@ -1095,8 -1071,9 +1167,9 @@@ } @VisibleForTesting - public void realMarkAlive(final InetAddress addr, final EndpointState localState) + public void realMarkAlive(final InetAddressAndPort addr, final EndpointState localState) { + checkProperThreadForStateMutation(); if (logger.isTraceEnabled()) logger.trace("marking as alive {}", addr); localState.markAlive(); @@@ -1115,8 -1090,9 +1188,9 @@@ } @VisibleForTesting - public void markDead(InetAddress addr, EndpointState localState) + public void markDead(InetAddressAndPort addr, EndpointState localState) { + checkProperThreadForStateMutation(); if (logger.isTraceEnabled()) logger.trace("marking as down {}", addr); localState.markDead(); @@@ -1137,8 -1111,9 +1211,9 @@@ * @param ep endpoint * @param epState EndpointState for the endpoint */ - private void handleMajorStateChange(InetAddress ep, EndpointState epState) + private void handleMajorStateChange(InetAddressAndPort ep, EndpointState epState) { + checkProperThreadForStateMutation(); EndpointState localEpState = endpointStateMap.get(ep); if (!isDeadState(epState)) { @@@ -1222,12 -1183,13 +1297,13 @@@ return pieces[0]; } - void applyStateLocally(Map<InetAddress, EndpointState> epStateMap) + void applyStateLocally(Map<InetAddressAndPort, EndpointState> epStateMap) { + checkProperThreadForStateMutation(); - for (Entry<InetAddress, EndpointState> entry : epStateMap.entrySet()) + for (Entry<InetAddressAndPort, EndpointState> entry : epStateMap.entrySet()) { - InetAddress ep = entry.getKey(); - if ( ep.equals(FBUtilities.getBroadcastAddress()) && !isInShadowRound()) + InetAddressAndPort ep = entry.getKey(); + if ( ep.equals(FBUtilities.getBroadcastAddressAndPort()) && !isInShadowRound()) continue; if (justRemovedEndpoints.containsKey(ep)) { @@@ -1649,9 -1554,10 +1725,10 @@@ /** * Add an endpoint we knew about previously, but whose state is unknown */ - public void addSavedEndpoint(InetAddress ep) + public void addSavedEndpoint(InetAddressAndPort ep) { + checkProperThreadForStateMutation(); - if (ep.equals(FBUtilities.getBroadcastAddress())) + if (ep.equals(FBUtilities.getBroadcastAddressAndPort())) { logger.debug("Attempt to add self as saved endpoint"); return; diff --cc src/java/org/apache/cassandra/service/StorageService.java index 6a57493,e64cbaa..2707b85 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@@ -1039,8 -1020,10 +1039,10 @@@ public class StorageService extends Not // remove the existing info about the replaced node. if (!current.isEmpty()) { - for (InetAddressAndPort existing : current) - Gossiper.instance.replacedEndpoint(existing); + Gossiper.runInGossipStageBlocking(() -> { - for (InetAddress existing : current) ++ for (InetAddressAndPort existing : current) + Gossiper.instance.replacedEndpoint(existing); + }); } } else @@@ -2756,9 -2609,9 +2758,9 @@@ } /** unlike excise we just need this endpoint gone without going through any notifications **/ - private void removeEndpoint(InetAddress endpoint) + private void removeEndpoint(InetAddressAndPort endpoint) { - Gossiper.instance.removeEndpoint(endpoint); + Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.removeEndpoint(endpoint)); SystemKeyspace.removeEndpoint(endpoint); } diff --cc test/distributed/org/apache/cassandra/distributed/impl/Instance.java index 00049f2,382388b..53e109a --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@@ -341,19 -349,21 +341,23 @@@ public class Instance extends IsolatedE for (int i = 0; i < tokens.size(); i++) { InetAddressAndPort ep = hosts.get(i); - Gossiper.instance.initializeNodeUnsafe(ep, hostIds.get(i), 1); - Gossiper.instance.injectApplicationState(ep, - ApplicationState.TOKENS, - new VersionedValue.VersionedValueFactory(partitioner).tokens(Collections.singleton(tokens.get(i)))); - storageService.onChange(ep, - ApplicationState.STATUS_WITH_PORT, - new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(tokens.get(i)))); - storageService.onChange(ep, - ApplicationState.STATUS, - new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(tokens.get(i)))); - Gossiper.instance.realMarkAlive(ep, Gossiper.instance.getEndpointStateForEndpoint(ep)); + UUID hostId = hostIds.get(i); + Token token = tokens.get(i); + Gossiper.runInGossipStageBlocking(() -> { - Gossiper.instance.initializeNodeUnsafe(ep.address, hostId, 1); - Gossiper.instance.injectApplicationState(ep.address, ++ Gossiper.instance.initializeNodeUnsafe(ep, hostId, 1); ++ Gossiper.instance.injectApplicationState(ep, + ApplicationState.TOKENS, + new VersionedValue.VersionedValueFactory(partitioner).tokens(Collections.singleton(token))); - storageService.onChange(ep.address, ++ storageService.onChange(ep, ++ ApplicationState.STATUS_WITH_PORT, ++ new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(token))); ++ storageService.onChange(ep, + ApplicationState.STATUS, + new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(token))); - Gossiper.instance.realMarkAlive(ep.address, Gossiper.instance.getEndpointStateForEndpoint(ep.address)); ++ Gossiper.instance.realMarkAlive(ep, Gossiper.instance.getEndpointStateForEndpoint(ep)); + }); - int version = Math.min(MessagingService.current_version, cluster.get(ep).getMessagingVersion()); - MessagingService.instance().setVersion(ep.address, version); + MessagingService.instance().setVersion(ep, version); } // check that all nodes are in token metadata diff --cc test/unit/org/apache/cassandra/gms/GossiperTest.java index a78d300,448620a..9c25b86 --- a/test/unit/org/apache/cassandra/gms/GossiperTest.java +++ b/test/unit/org/apache/cassandra/gms/GossiperTest.java @@@ -47,9 -47,15 +47,10 @@@ import static org.junit.Assert.assertTr public class GossiperTest { - @BeforeClass - public static void before() + static { + System.setProperty(Gossiper.Props.DISABLE_THREAD_VALIDATION, "true"); DatabaseDescriptor.daemonInitialization(); - SchemaLoader.prepareServer(); - SchemaLoader.createKeyspace("schema_test_ks", - KeyspaceParams.simple(1), - SchemaLoader.standardCFMD("schema_test_ks", "schema_test_cf")); } static final IPartitioner partitioner = new RandomPartitioner(); diff --cc test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java index 1645d77,0000000..af72456 mode 100644,000000..100644 --- a/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java +++ b/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java @@@ -1,300 -1,0 +1,300 @@@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.net; + +import java.net.UnknownHostException; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.config.DatabaseDescriptor; +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 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() + { + DatabaseDescriptor.daemonInitialization(); + } + + @Before + public void setUp() throws UnknownHostException + { + 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.addAll(peersA); + peers.addAll(peersB); + peers.addAll(peersC); + } + + @After + public void tearDown() + { + MessagingService.instance().clearMessageSinks(); + } + + @Test + public void execute_HappyPath() + { + Sink sink = new Sink(true, true, peers); + MessagingService.instance().addMessageSink(sink); + Assert.assertTrue(localQuorumConnectivityChecker.execute(peers, this::getDatacenter)); + Assert.assertTrue(checkAllConnectionTypesSeen(sink)); + } + + @Test + public void execute_NotAlive() + { + Sink sink = new Sink(false, true, peers); + MessagingService.instance().addMessageSink(sink); + Assert.assertFalse(localQuorumConnectivityChecker.execute(peers, this::getDatacenter)); + Assert.assertTrue(checkAllConnectionTypesSeen(sink)); + } + + @Test + public void execute_NoConnectionsAcks() + { + 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.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 boolean checkAllConnectionTypesSeen(Sink sink) + { + boolean result = true; + for (InetAddressAndPort peer : peers) + { + if (peer.equals(FBUtilities.getBroadcastAddressAndPort())) + continue; + ConnectionTypeRecorder recorder = sink.seenConnectionRequests.get(peer); + 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, Set<InetAddressAndPort> aliveHosts) + { + this.markAliveInGossip = markAliveInGossip; + this.processConnectAck = processConnectAck; + this.aliveHosts = aliveHosts; + seenConnectionRequests = new HashMap<>(); + } + + @Override + public boolean allowOutgoingMessage(MessageOut message, int id, InetAddressAndPort to) + { + ConnectionTypeRecorder recorder = seenConnectionRequests.computeIfAbsent(to, inetAddress -> new ConnectionTypeRecorder()); + if (message.connectionType == SMALL_MESSAGE) + { + Assert.assertFalse(recorder.seenSmallMessageRequest); + recorder.seenSmallMessageRequest = true; + } + else + { + Assert.assertFalse(recorder.seenLargeMessageRequest); + 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); + MessagingService.instance().getRegisteredCallback(id).callback.response(msgIn); + } + + if (markAliveInGossip) - Gossiper.instance.realMarkAlive(to, new EndpointState(new HeartBeatState(1, 1))); ++ Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.realMarkAlive(to, new EndpointState(new HeartBeatState(1, 1)))); + return false; + } + + @Override + public boolean allowIncomingMessage(MessageIn message, int id) + { + return false; + } + } + + private static class ConnectionTypeRecorder + { + boolean seenSmallMessageRequest; + boolean seenLargeMessageRequest; + } +} diff --cc test/unit/org/apache/cassandra/service/StorageServiceServerTest.java index e4ecbea,297d19d..2db221b --- a/test/unit/org/apache/cassandra/service/StorageServiceServerTest.java +++ b/test/unit/org/apache/cassandra/service/StorageServiceServerTest.java @@@ -69,6 -64,6 +69,7 @@@ public class StorageServiceServerTes @BeforeClass public static void setUp() throws ConfigurationException { ++ System.setProperty(Gossiper.Props.DISABLE_THREAD_VALIDATION, "true"); DatabaseDescriptor.daemonInitialization(); IEndpointSnitch snitch = new PropertyFileSnitch(); DatabaseDescriptor.setEndpointSnitch(snitch); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
