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

Merge branch 'cassandra-3.0' into cassandra-3.11

This commit is contained in:
Blake Eggleston 2019-04-25 10:21:22 -07:00
parent 839fc7e36f c3ce32e239
commit 01d6548e0d
13 changed files with 170 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -2,6 +2,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)

1
build.xml
View File

@ -1274,6 +1274,7 @@
<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/>
<classpath>
<pathelement path="${java.class.path}"/>

2
ide/idea/workspace.xml
View File

@ -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 -ea -XX:MaxMetaspaceSize=256M -XX:SoftRefLRUPolicyMSPerMB=0" />
<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="PARAMETERS" value="" />
<option name="WORKING_DIRECTORY" value="" />
<option name="ENV_VARIABLES" />

8
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -127,6 +127,9 @@ public class DatabaseDescriptor
private static final boolean disableSTCSInL0 = Boolean.getBoolean(Config.PROPERTY_PREFIX + "disable_stcs_in_l0");
private static final boolean unsafeSystem = Boolean.getBoolean(Config.PROPERTY_PREFIX + "unsafesystem");
// turns some warnings into exceptions for testing
private static final boolean strictRuntimeChecks = Boolean.getBoolean("cassandra.strict.runtime.checks");
public static void daemonInitialization() throws ConfigurationException
{
daemonInitialization(DatabaseDescriptor::loadConfig);
@ -2529,4 +2532,9 @@ public class DatabaseDescriptor
{
return backPressureStrategy;
}
public static boolean strictRuntimeChecks()
{
return strictRuntimeChecks;
}
}

192
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -27,11 +27,15 @@ import java.util.concurrent.locks.ReentrantLock;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.ListenableFutureTask;
import com.google.common.util.concurrent.Uninterruptibles;
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;
@ -63,11 +67,17 @@ import org.apache.cassandra.utils.JVMStabilityInspector;
* of the three above mentioned messages updates the Failure Detector with the liveness information.
* Upon hearing a GossipShutdownMessage, this module will instantly mark the remote node as down in
* the Failure Detector.
*
* This class is not threadsafe and any state changes should happen in the gossip stage.
*/
public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
{
public static final String MBEAN_NAME = "org.apache.cassandra.net:type=Gossiper";
public static class Props
{
public static final String DISABLE_THREAD_VALIDATION = "cassandra.gossip.disable_thread_validation";
}
private static final DebuggableScheduledThreadPoolExecutor executor = new DebuggableScheduledThreadPoolExecutor("GossipTasks");
@ -88,6 +98,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
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();
// Timestamp to prevent processing any in-flight messages for we've not send any SYN yet, see CASSANDRA-12653.
@ -140,6 +151,37 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
private volatile long lastProcessedMessageAt = System.currentTimeMillis();
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()
@ -327,6 +369,27 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
return state.equals(VersionedValue.SHUTDOWN);
}
public static void runInGossipStageBlocking(Runnable runnable)
{
// run immediately if we're already in the gossip stage
if (isInGossipStage())
{
runnable.run();
return;
}
ListenableFutureTask task = ListenableFutureTask.create(runnable, null);
StageManager.getStage(Stage.GOSSIP).execute(task);
try
{
task.get();
}
catch (InterruptedException | ExecutionException e)
{
throw new AssertionError(e);
}
}
/**
* This method is part of IFailureDetectionEventListener interface. This is invoked
* by the Failure Detector when it convicts an end point.
@ -335,24 +398,26 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
public void convict(InetAddress endpoint, double phi)
{
EndpointState epState = endpointStateMap.get(endpoint);
if (epState == null)
return;
runInGossipStageBlocking(() -> {
EndpointState epState = endpointStateMap.get(endpoint);
if (epState == null)
return;
if (!epState.isAlive())
return;
if (!epState.isAlive())
return;
logger.debug("Convicting {} with status {} - alive {}", endpoint, getGossipStatus(epState), epState.isAlive());
logger.debug("Convicting {} with status {} - alive {}", endpoint, getGossipStatus(epState), epState.isAlive());
if (isShutdown(endpoint))
{
markAsShutdown(endpoint);
}
else
{
markDead(endpoint, epState);
}
if (isShutdown(endpoint))
{
markAsShutdown(endpoint);
}
else
{
markDead(endpoint, epState);
}
});
}
/**
@ -361,6 +426,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
protected void markAsShutdown(InetAddress endpoint)
{
checkProperThreadForStateMutation();
EndpointState epState = endpointStateMap.get(endpoint);
if (epState == null)
return;
@ -392,6 +458,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
private void evictFromMembership(InetAddress endpoint)
{
checkProperThreadForStateMutation();
unreachableEndpoints.remove(endpoint);
endpointStateMap.remove(endpoint);
expireTimeEndpointMap.remove(endpoint);
@ -406,6 +473,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
public void removeEndpoint(InetAddress 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);
@ -466,6 +534,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
public void replacedEndpoint(InetAddress endpoint)
{
checkProperThreadForStateMutation();
removeEndpoint(endpoint);
evictFromMembership(endpoint);
replacementQuarantine(endpoint);
@ -578,49 +647,51 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
public void assassinateEndpoint(String address) throws UnknownHostException
{
InetAddress endpoint = InetAddress.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
epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.left(tokens, computeExpireTime()));
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
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)
@ -802,8 +873,10 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
&& TimeUnit.NANOSECONDS.toMillis(nowNano - epState.getUpdateTimestamp()) > fatClientTimeout)
{
logger.info("FatClient {} has been silent for {}ms, removing from gossip", endpoint, fatClientTimeout);
removeEndpoint(endpoint); // will put it in justRemovedEndpoints to respect quarantine delay
evictFromMembership(endpoint); // can get rid of the state immediately
runInGossipStageBlocking(() -> {
removeEndpoint(endpoint); // will put it in justRemovedEndpoints to respect quarantine delay
evictFromMembership(endpoint); // can get rid of the state immediately
});
}
// check for dead state removal
@ -990,7 +1063,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
public void response(MessageIn msg)
{
realMarkAlive(addr, localState);
runInGossipStageBlocking(() -> realMarkAlive(addr, localState));
}
};
@ -1000,6 +1073,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
@VisibleForTesting
public void realMarkAlive(final InetAddress addr, final EndpointState localState)
{
checkProperThreadForStateMutation();
if (logger.isTraceEnabled())
logger.trace("marking as alive {}", addr);
localState.markAlive();
@ -1018,6 +1092,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
@VisibleForTesting
public void markDead(InetAddress addr, EndpointState localState)
{
checkProperThreadForStateMutation();
if (logger.isTraceEnabled())
logger.trace("marking as down {}", addr);
localState.markDead();
@ -1038,6 +1113,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
private void handleMajorStateChange(InetAddress ep, EndpointState epState)
{
checkProperThreadForStateMutation();
EndpointState localEpState = endpointStateMap.get(ep);
if (!isDeadState(epState))
{
@ -1109,6 +1185,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
void applyStateLocally(Map<InetAddress, EndpointState> epStateMap)
{
checkProperThreadForStateMutation();
for (Entry<InetAddress, EndpointState> entry : epStateMap.entrySet())
{
InetAddress ep = entry.getKey();
@ -1479,6 +1556,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
public void addSavedEndpoint(InetAddress ep)
{
checkProperThreadForStateMutation();
if (ep.equals(FBUtilities.getBroadcastAddress()))
{
logger.debug("Attempt to add self as saved endpoint");

10
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -734,7 +734,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
{
if (loadedHostIds.containsKey(ep))
tokenMetadata.updateHostId(loadedHostIds.get(ep), ep);
Gossiper.instance.addSavedEndpoint(ep);
Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.addSavedEndpoint(ep));
}
}
}
@ -1020,8 +1020,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
// remove the existing info about the replaced node.
if (!current.isEmpty())
{
for (InetAddress existing : current)
Gossiper.instance.replacedEndpoint(existing);
Gossiper.runInGossipStageBlocking(() -> {
for (InetAddress existing : current)
Gossiper.instance.replacedEndpoint(existing);
});
}
}
else
@ -2609,7 +2611,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
/** unlike excise we just need this endpoint gone without going through any notifications **/
private void removeEndpoint(InetAddress endpoint)
{
Gossiper.instance.removeEndpoint(endpoint);
Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.removeEndpoint(endpoint));
SystemKeyspace.removeEndpoint(endpoint);
}

20
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -349,14 +349,18 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
for (int i = 0; i < tokens.size(); i++)
{
InetAddressAndPort ep = hosts.get(i);
Gossiper.instance.initializeNodeUnsafe(ep.address, hostIds.get(i), 1);
Gossiper.instance.injectApplicationState(ep.address,
ApplicationState.TOKENS,
new VersionedValue.VersionedValueFactory(partitioner).tokens(Collections.singleton(tokens.get(i))));
storageService.onChange(ep.address,
ApplicationState.STATUS,
new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(tokens.get(i))));
Gossiper.instance.realMarkAlive(ep.address, Gossiper.instance.getEndpointStateForEndpoint(ep.address));
UUID hostId = hostIds.get(i);
Token token = tokens.get(i);
Gossiper.runInGossipStageBlocking(() -> {
Gossiper.instance.initializeNodeUnsafe(ep.address, hostId, 1);
Gossiper.instance.injectApplicationState(ep.address,
ApplicationState.TOKENS,
new VersionedValue.VersionedValueFactory(partitioner).tokens(Collections.singleton(token)));
storageService.onChange(ep.address,
ApplicationState.STATUS,
new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(token)));
Gossiper.instance.realMarkAlive(ep.address, Gossiper.instance.getEndpointStateForEndpoint(ep.address));
});
int version = Math.min(MessagingService.current_version, cluster.get(ep).getMessagingVersion());
MessagingService.instance().setVersion(ep.address, version);

1
test/unit/org/apache/cassandra/gms/GossiperTest.java
View File

@ -50,6 +50,7 @@ public class GossiperTest
@BeforeClass
public static void before()
{
System.setProperty(Gossiper.Props.DISABLE_THREAD_VALIDATION, "true");
DatabaseDescriptor.daemonInitialization();
SchemaLoader.prepareServer();
SchemaLoader.createKeyspace("schema_test_ks",

1
test/unit/org/apache/cassandra/gms/PendingRangeCalculatorServiceTest.java
View File

@ -55,6 +55,7 @@ public class PendingRangeCalculatorServiceTest
@BeforeClass
public static void setUp() throws ConfigurationException
{
System.setProperty(Gossiper.Props.DISABLE_THREAD_VALIDATION, "true");
SchemaLoader.prepareServer();
StorageService.instance.initServer();
}

1
test/unit/org/apache/cassandra/locator/CloudstackSnitchTest.java
View File

@ -45,6 +45,7 @@ public class CloudstackSnitchTest
@BeforeClass
public static void setup() throws Exception
{
System.setProperty(Gossiper.Props.DISABLE_THREAD_VALIDATION, "true");
DatabaseDescriptor.daemonInitialization();
SchemaLoader.mkdirs();
SchemaLoader.cleanup();

1
test/unit/org/apache/cassandra/locator/EC2SnitchTest.java
View File

@ -50,6 +50,7 @@ public class EC2SnitchTest
@BeforeClass
public static void setup() throws Exception
{
System.setProperty(Gossiper.Props.DISABLE_THREAD_VALIDATION, "true");
DatabaseDescriptor.daemonInitialization();
SchemaLoader.mkdirs();
SchemaLoader.cleanup();

1
test/unit/org/apache/cassandra/locator/GoogleCloudSnitchTest.java
View File

@ -46,6 +46,7 @@ public class GoogleCloudSnitchTest
@BeforeClass
public static void setup() throws Exception
{
System.setProperty(Gossiper.Props.DISABLE_THREAD_VALIDATION, "true");
DatabaseDescriptor.daemonInitialization();
SchemaLoader.mkdirs();
SchemaLoader.cleanup();

1
test/unit/org/apache/cassandra/locator/PropertyFileSnitchTest.java
View File

@ -72,6 +72,7 @@ public class PropertyFileSnitchTest
@Before
public void setup() throws ConfigurationException, IOException
{
System.setProperty(Gossiper.Props.DISABLE_THREAD_VALIDATION, "true");
String confFile = FBUtilities.resourceToFile(PropertyFileSnitch.SNITCH_PROPERTIES_FILENAME);
effectiveFile = Paths.get(confFile);
backupFile = Paths.get(confFile + ".bak");