mirror of https://github.com/apache/cassandra
Improve unit tests performance
- Don't flush schema on every schema update in unit tests - Use unix command to delete test data - Shorten teardown - Stable processor count presented by JMX on Jenkins, CircleCI and local Patch by <jacek-lewandowski>, reviewed by <michaelsembwever> and <josh-mckenzie> for CASSANDRA-17427
This commit is contained in:
parent
ef2066fe60
commit
9bf128aaa3
|
|
@ -1,4 +1,5 @@
|
||||||
4.0.8
|
4.0.8
|
||||||
|
* Improve unit tests performance (CASSANDRA-17427)
|
||||||
* Connect to listen address when own broadcast address is requested (CASSANDRA-18200)
|
* Connect to listen address when own broadcast address is requested (CASSANDRA-18200)
|
||||||
* Add safeguard so cleanup fails when node has pending ranges (CASSANDRA-16418)
|
* Add safeguard so cleanup fails when node has pending ranges (CASSANDRA-16418)
|
||||||
* Fix legacy clustering serialization for paging with compact storage (CASSANDRA-17507)
|
* Fix legacy clustering serialization for paging with compact storage (CASSANDRA-17507)
|
||||||
|
|
|
||||||
17
build.xml
17
build.xml
|
|
@ -111,12 +111,18 @@
|
||||||
<property name="test.burn.timeout" value="60000000" />
|
<property name="test.burn.timeout" value="60000000" />
|
||||||
<property name="test.distributed.timeout" value="900000" />
|
<property name="test.distributed.timeout" value="900000" />
|
||||||
|
|
||||||
<!-- default for cql tests. Can be override by -Dcassandra.test.use_prepared=false -->
|
<!-- default for cql tests. Can be overridden by -Dcassandra.test.use_prepared=false -->
|
||||||
<property name="cassandra.test.use_prepared" value="true" />
|
<property name="cassandra.test.use_prepared" value="true" />
|
||||||
|
|
||||||
|
<!-- The number of active processors seen by JVM -->
|
||||||
|
<property name="cassandra.test.processorCount" value="4"/>
|
||||||
|
|
||||||
<!-- skip flushing schema tables during tests -->
|
<!-- skip flushing schema tables during tests -->
|
||||||
<property name="cassandra.test.flush_local_schema_changes" value="false" />
|
<property name="cassandra.test.flush_local_schema_changes" value="false" />
|
||||||
|
|
||||||
|
<!-- fast shutdown of messaging service -->
|
||||||
|
<property name="cassandra.test.messagingService.nonGracefulShutdown" value="true"/>
|
||||||
|
|
||||||
<!-- https://www.eclemma.org/jacoco/ -->
|
<!-- https://www.eclemma.org/jacoco/ -->
|
||||||
<property name="jacoco.export.dir" value="${build.dir}/jacoco/" />
|
<property name="jacoco.export.dir" value="${build.dir}/jacoco/" />
|
||||||
<property name="jacoco.partials.dir" value="${jacoco.export.dir}/partials" />
|
<property name="jacoco.partials.dir" value="${jacoco.export.dir}/partials" />
|
||||||
|
|
@ -152,6 +158,10 @@
|
||||||
<available file="${build.src.java}" type="dir" />
|
<available file="${build.src.java}" type="dir" />
|
||||||
</condition>
|
</condition>
|
||||||
|
|
||||||
|
<condition property="cassandra.use_nix_recursive_delete" value="false" else="true">
|
||||||
|
<os family="windows" />
|
||||||
|
</condition>
|
||||||
|
|
||||||
<!-- Check if all tests are being run or just one. If it's all tests don't spam the console with test output.
|
<!-- Check if all tests are being run or just one. If it's all tests don't spam the console with test output.
|
||||||
If it's an individual test print the output from the test under the assumption someone is debugging the test
|
If it's an individual test print the output from the test under the assumption someone is debugging the test
|
||||||
and wants to know what is going on without having to context switch to the log file that is generated.
|
and wants to know what is going on without having to context switch to the log file that is generated.
|
||||||
|
|
@ -1406,6 +1416,7 @@
|
||||||
more aggressively rather than waiting. See CASSANDRA-14922 for more details.
|
more aggressively rather than waiting. See CASSANDRA-14922 for more details.
|
||||||
-->
|
-->
|
||||||
<jvmarg value="-XX:SoftRefLRUPolicyMSPerMB=0" />
|
<jvmarg value="-XX:SoftRefLRUPolicyMSPerMB=0" />
|
||||||
|
<jvmarg value="-XX:ActiveProcessorCount=${cassandra.test.processorCount}" />
|
||||||
<jvmarg value="-Dcassandra.test.driver.connection_timeout_ms=${test.driver.connection_timeout_ms}"/>
|
<jvmarg value="-Dcassandra.test.driver.connection_timeout_ms=${test.driver.connection_timeout_ms}"/>
|
||||||
<jvmarg value="-Dcassandra.test.driver.read_timeout_ms=${test.driver.read_timeout_ms}"/>
|
<jvmarg value="-Dcassandra.test.driver.read_timeout_ms=${test.driver.read_timeout_ms}"/>
|
||||||
<jvmarg value="-Dcassandra.memtable_row_overhead_computation_step=100"/>
|
<jvmarg value="-Dcassandra.memtable_row_overhead_computation_step=100"/>
|
||||||
|
|
@ -1416,6 +1427,9 @@
|
||||||
<jvmarg value="-Dcassandra.testtag=@{testtag}"/>
|
<jvmarg value="-Dcassandra.testtag=@{testtag}"/>
|
||||||
<jvmarg value="-Dcassandra.keepBriefBrief=${cassandra.keepBriefBrief}" />
|
<jvmarg value="-Dcassandra.keepBriefBrief=${cassandra.keepBriefBrief}" />
|
||||||
<jvmarg value="-Dcassandra.strict.runtime.checks=true" />
|
<jvmarg value="-Dcassandra.strict.runtime.checks=true" />
|
||||||
|
<jvmarg value="-Dcassandra.test.flush_local_schema_changes=${cassandra.test.flush_local_schema_changes}"/>
|
||||||
|
<jvmarg value="-Dcassandra.test.messagingService.nonGracefulShutdown=${cassandra.test.messagingService.nonGracefulShutdown}"/>
|
||||||
|
<jvmarg value="-Dcassandra.use_nix_recursive_delete=${cassandra.use_nix_recursive_delete}"/>
|
||||||
<jvmarg line="${java11-jvmargs}"/>
|
<jvmarg line="${java11-jvmargs}"/>
|
||||||
<!-- disable shrinks in quicktheories CASSANDRA-15554 -->
|
<!-- disable shrinks in quicktheories CASSANDRA-15554 -->
|
||||||
<jvmarg value="-DQT_SHRINKS=0"/>
|
<jvmarg value="-DQT_SHRINKS=0"/>
|
||||||
|
|
@ -2009,6 +2023,7 @@
|
||||||
<fileset dir="ide/idea"/>
|
<fileset dir="ide/idea"/>
|
||||||
</copy>
|
</copy>
|
||||||
<replace file=".idea/workspace.xml" token="trunk" value="${eclipse.project.name}"/>
|
<replace file=".idea/workspace.xml" token="trunk" value="${eclipse.project.name}"/>
|
||||||
|
<replace file=".idea/workspace.xml" token="-Dcassandra.use_nix_recursive_delete=true" value="-Dcassandra.use_nix_recursive_delete=${cassandra.use_nix_recursive_delete}"/>
|
||||||
<copy tofile="${eclipse.project.name}.iml" file="ide/idea-iml-file.xml"/>
|
<copy tofile="${eclipse.project.name}.iml" file="ide/idea-iml-file.xml"/>
|
||||||
<echo file=".idea/.name">Apache Cassandra ${eclipse.project.name}</echo>
|
<echo file=".idea/.name">Apache Cassandra ${eclipse.project.name}</echo>
|
||||||
<echo file=".idea/modules.xml"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
|
<echo file=".idea/modules.xml"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
|
||||||
|
|
|
||||||
|
|
@ -161,13 +161,19 @@
|
||||||
<configuration default="true" type="JUnit" factoryName="JUnit">
|
<configuration default="true" type="JUnit" factoryName="JUnit">
|
||||||
<extension name="coverage" enabled="false" merge="false" sample_coverage="true" runner="idea" />
|
<extension name="coverage" enabled="false" merge="false" sample_coverage="true" runner="idea" />
|
||||||
<module name="" />
|
<module name="" />
|
||||||
|
<extension name="coverage">
|
||||||
|
<pattern>
|
||||||
|
<option name="PATTERN" value="org.apache.cassandra.*" />
|
||||||
|
<option name="ENABLED" value="true" />
|
||||||
|
</pattern>
|
||||||
|
</extension>
|
||||||
<option name="ALTERNATIVE_JRE_PATH_ENABLED" value="false" />
|
<option name="ALTERNATIVE_JRE_PATH_ENABLED" value="false" />
|
||||||
<option name="ALTERNATIVE_JRE_PATH" value="" />
|
<option name="ALTERNATIVE_JRE_PATH" value="" />
|
||||||
<option name="PACKAGE_NAME" />
|
<option name="PACKAGE_NAME" />
|
||||||
<option name="MAIN_CLASS_NAME" value="" />
|
<option name="MAIN_CLASS_NAME" value="" />
|
||||||
<option name="METHOD_NAME" value="" />
|
<option name="METHOD_NAME" value="" />
|
||||||
<option name="TEST_OBJECT" value="class" />
|
<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:MaxMetaspaceSize=384M -XX:SoftRefLRUPolicyMSPerMB=0 -Dcassandra.strict.runtime.checks=true -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dmigration-sstable-root=$PROJECT_DIR$/test/data/migration-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.tolerate_sstable_size=true -Dcassandra.skip_sync=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:MaxMetaspaceSize=384M -XX:SoftRefLRUPolicyMSPerMB=0 -XX:ActiveProcessorCount=4 -Dcassandra.strict.runtime.checks=true -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dmigration-sstable-root=$PROJECT_DIR$/test/data/migration-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.tolerate_sstable_size=true -Dcassandra.skip_sync=true -Dcassandra.use_nix_recursive_delete=true -Dcassandra.test.messagingService.nonGracefulShutdown=true -Dcassandra.test.flush_local_schema_changes=false " />
|
||||||
<option name="PARAMETERS" value="" />
|
<option name="PARAMETERS" value="" />
|
||||||
<fork_mode value="class" />
|
<fork_mode value="class" />
|
||||||
<option name="WORKING_DIRECTORY" value="" />
|
<option name="WORKING_DIRECTORY" value="" />
|
||||||
|
|
|
||||||
|
|
@ -204,7 +204,24 @@ public enum CassandraRelevantProperties
|
||||||
* immediately as it is high chance they will fail anyway. It is better to wait a bit instead of flooding logs
|
* immediately as it is high chance they will fail anyway. It is better to wait a bit instead of flooding logs
|
||||||
* and wasting resources.
|
* and wasting resources.
|
||||||
*/
|
*/
|
||||||
SCHEMA_PULL_BACKOFF_DELAY_MS("cassandra.schema_pull_backoff_delay_ms", "3000");
|
SCHEMA_PULL_BACKOFF_DELAY_MS("cassandra.schema_pull_backoff_delay_ms", "3000"),
|
||||||
|
|
||||||
|
/** When enabled, recursive directory deletion will be executed using a unix command `rm -rf` instead of traversing
|
||||||
|
* and removing individual files. This is now used only tests, but eventually we will make it true by default.*/
|
||||||
|
USE_NIX_RECURSIVE_DELETE("cassandra.use_nix_recursive_delete"),
|
||||||
|
|
||||||
|
/** If set, {@link org.apache.cassandra.net.MessagingService} is shutdown abrtuptly without waiting for anything.
|
||||||
|
* This is an optimization used in unit tests becuase we never restart a node there. The only node is stopoped
|
||||||
|
* when the JVM terminates. Therefore, we can use such optimization and not wait unnecessarily. */
|
||||||
|
NON_GRACEFUL_SHUTDOWN("cassandra.test.messagingService.nonGracefulShutdown"),
|
||||||
|
|
||||||
|
/** Flush changes of {@link org.apache.cassandra.schema.SchemaKeyspace} after each schema modification. In production,
|
||||||
|
* we always do that. However, tests which do not restart nodes may disable this functionality in order to run
|
||||||
|
* faster. Note that this is disabled for unit tests but if an individual test requires schema to be flushed, it
|
||||||
|
* can be also done manually for that particular case: {@code flush(SchemaConstants.SCHEMA_KEYSPACE_NAME);}. */
|
||||||
|
FLUSH_LOCAL_SCHEMA_CHANGES("cassandra.test.flush_local_schema_changes", "true"),
|
||||||
|
|
||||||
|
;
|
||||||
|
|
||||||
CassandraRelevantProperties(String key, String defaultVal)
|
CassandraRelevantProperties(String key, String defaultVal)
|
||||||
{
|
{
|
||||||
|
|
@ -237,6 +254,17 @@ public enum CassandraRelevantProperties
|
||||||
return value == null ? defaultVal : STRING_CONVERTER.convert(value);
|
return value == null ? defaultVal : STRING_CONVERTER.convert(value);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Sets the property to its default value if a default value was specified. Remove the property otherwise.
|
||||||
|
*/
|
||||||
|
public void reset()
|
||||||
|
{
|
||||||
|
if (defaultVal != null)
|
||||||
|
System.setProperty(key, defaultVal);
|
||||||
|
else
|
||||||
|
System.getProperties().remove(key);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Gets the value of a system property as a String.
|
* Gets the value of a system property as a String.
|
||||||
* @return system property String value if it exists, overrideDefaultValue otherwise.
|
* @return system property String value if it exists, overrideDefaultValue otherwise.
|
||||||
|
|
@ -355,4 +383,3 @@ public enum CassandraRelevantProperties
|
||||||
return System.getProperties().containsKey(key);
|
return System.getProperties().containsKey(key);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -57,12 +57,15 @@ import org.apache.cassandra.io.FSReadError;
|
||||||
import org.apache.cassandra.io.FSWriteError;
|
import org.apache.cassandra.io.FSWriteError;
|
||||||
import org.apache.cassandra.io.sstable.CorruptSSTableException;
|
import org.apache.cassandra.io.sstable.CorruptSSTableException;
|
||||||
import org.apache.cassandra.service.StorageService;
|
import org.apache.cassandra.service.StorageService;
|
||||||
|
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||||
|
import org.apache.cassandra.utils.FBUtilities;
|
||||||
import org.apache.cassandra.utils.JVMStabilityInspector;
|
import org.apache.cassandra.utils.JVMStabilityInspector;
|
||||||
import org.apache.cassandra.utils.NoSpamLogger;
|
import org.apache.cassandra.utils.NoSpamLogger;
|
||||||
import org.apache.cassandra.utils.SyncUtil;
|
import org.apache.cassandra.utils.SyncUtil;
|
||||||
|
|
||||||
import static com.google.common.base.Throwables.propagate;
|
import static com.google.common.base.Throwables.propagate;
|
||||||
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_IO_TMPDIR;
|
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_IO_TMPDIR;
|
||||||
|
import static org.apache.cassandra.config.CassandraRelevantProperties.USE_NIX_RECURSIVE_DELETE;
|
||||||
import static org.apache.cassandra.utils.Throwables.maybeFail;
|
import static org.apache.cassandra.utils.Throwables.maybeFail;
|
||||||
import static org.apache.cassandra.utils.Throwables.merge;
|
import static org.apache.cassandra.utils.Throwables.merge;
|
||||||
|
|
||||||
|
|
@ -671,9 +674,15 @@ public final class FileUtils
|
||||||
*/
|
*/
|
||||||
public static void deleteRecursive(File dir)
|
public static void deleteRecursive(File dir)
|
||||||
{
|
{
|
||||||
|
if (USE_NIX_RECURSIVE_DELETE.getBoolean() && dir.toPath().getFileSystem() == FileSystems.getDefault())
|
||||||
|
{
|
||||||
|
deleteRecursiveUsingNixCommand(dir.toPath(), false);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
deleteChildrenRecursive(dir);
|
deleteChildrenRecursive(dir);
|
||||||
|
|
||||||
// The directory is now empty so now it can be smoked
|
// The directory is now empty, so now it can be smoked
|
||||||
deleteWithConfirm(dir);
|
deleteWithConfirm(dir);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -688,8 +697,64 @@ public final class FileUtils
|
||||||
if (dir.isDirectory())
|
if (dir.isDirectory())
|
||||||
{
|
{
|
||||||
String[] children = dir.list();
|
String[] children = dir.list();
|
||||||
for (String child : children)
|
if (children.length == 0)
|
||||||
deleteRecursive(new File(dir, child));
|
return;
|
||||||
|
|
||||||
|
if (USE_NIX_RECURSIVE_DELETE.getBoolean() && dir.toPath().getFileSystem() == FileSystems.getDefault())
|
||||||
|
{
|
||||||
|
for (String child : children)
|
||||||
|
deleteRecursiveUsingNixCommand(dir.toPath().resolve(child), false);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
for (String child : children)
|
||||||
|
deleteRecursive(new File(dir, child));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Uses unix `rm -r` to delete a directory recursively.
|
||||||
|
* This method can be much faster than deleting files and directories recursively by traversing them with Java.
|
||||||
|
* Though, we use it only for tests because it provides less information about the problem when something goes wrong.
|
||||||
|
*
|
||||||
|
* @param path path to be deleted
|
||||||
|
* @param quietly if quietly, additional `-f` flag is added to the `rm` command so that it will not complain in case
|
||||||
|
* the provided path is missing
|
||||||
|
*/
|
||||||
|
private static void deleteRecursiveUsingNixCommand(Path path, boolean quietly)
|
||||||
|
{
|
||||||
|
String[] cmd = new String[]{ "rm",
|
||||||
|
quietly ? "-drf" : "-dr",
|
||||||
|
path.toAbsolutePath().toString() };
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
Process p = Runtime.getRuntime().exec(cmd);
|
||||||
|
int result = p.waitFor();
|
||||||
|
|
||||||
|
String out, err;
|
||||||
|
try (BufferedReader outReader = new BufferedReader(new InputStreamReader(p.getInputStream()));
|
||||||
|
BufferedReader errReader = new BufferedReader(new InputStreamReader(p.getErrorStream())))
|
||||||
|
{
|
||||||
|
out = outReader.lines().collect(Collectors.joining("\n"));
|
||||||
|
err = errReader.lines().collect(Collectors.joining("\n"));
|
||||||
|
}
|
||||||
|
|
||||||
|
if (result != 0 && Files.exists(path))
|
||||||
|
{
|
||||||
|
logger.error("{} returned:\nstdout:\n{}\n\nstderr:\n{}", Arrays.toString(cmd), out, err);
|
||||||
|
throw new IOException(String.format("%s returned non-zero exit code: %d%nstdout:%n%s%n%nstderr:%n%s", Arrays.toString(cmd), result, out, err));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (IOException e)
|
||||||
|
{
|
||||||
|
throw new FSWriteError(e, path.toString());
|
||||||
|
}
|
||||||
|
catch (InterruptedException e)
|
||||||
|
{
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
throw new FSWriteError(e, path.toString());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -44,6 +44,7 @@ import org.apache.cassandra.utils.FBUtilities;
|
||||||
import static java.util.Collections.synchronizedList;
|
import static java.util.Collections.synchronizedList;
|
||||||
import static java.util.concurrent.TimeUnit.MINUTES;
|
import static java.util.concurrent.TimeUnit.MINUTES;
|
||||||
import static org.apache.cassandra.concurrent.Stage.MUTATION;
|
import static org.apache.cassandra.concurrent.Stage.MUTATION;
|
||||||
|
import static org.apache.cassandra.config.CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN;
|
||||||
import static org.apache.cassandra.utils.Throwables.maybeFail;
|
import static org.apache.cassandra.utils.Throwables.maybeFail;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -422,15 +423,20 @@ public final class MessagingService extends MessagingServiceMBeanImpl
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Wait for callbacks and don't allow any more to be created (since they could require writing hints)
|
* Wait for callbacks and don't allow anymore to be created (since they could require writing hints)
|
||||||
*/
|
*/
|
||||||
public void shutdown()
|
public void shutdown()
|
||||||
{
|
{
|
||||||
shutdown(1L, MINUTES, true, true);
|
if (NON_GRACEFUL_SHUTDOWN.getBoolean())
|
||||||
|
// this branch is used in unit-tests when we really never restart a node and shutting down means the end of test
|
||||||
|
shutdownAbrubtly();
|
||||||
|
else
|
||||||
|
shutdown(1L, MINUTES, true, true);
|
||||||
}
|
}
|
||||||
|
|
||||||
public void shutdown(long timeout, TimeUnit units, boolean shutdownGracefully, boolean shutdownExecutors)
|
public void shutdown(long timeout, TimeUnit units, boolean shutdownGracefully, boolean shutdownExecutors)
|
||||||
{
|
{
|
||||||
|
logger.debug("Shutting down: timeout={}s, gracefully={}, shutdownExecutors={}", units.toSeconds(timeout), shutdownGracefully, shutdownExecutors);
|
||||||
if (isShuttingDown)
|
if (isShuttingDown)
|
||||||
{
|
{
|
||||||
logger.info("Shutdown was already called");
|
logger.info("Shutdown was already called");
|
||||||
|
|
@ -454,7 +460,7 @@ public final class MessagingService extends MessagingServiceMBeanImpl
|
||||||
() -> {
|
() -> {
|
||||||
List<ExecutorService> inboundExecutors = new ArrayList<>();
|
List<ExecutorService> inboundExecutors = new ArrayList<>();
|
||||||
inboundSockets.close(synchronizedList(inboundExecutors)::add).get();
|
inboundSockets.close(synchronizedList(inboundExecutors)::add).get();
|
||||||
ExecutorUtils.awaitTermination(1L, TimeUnit.MINUTES, inboundExecutors);
|
ExecutorUtils.awaitTermination(timeout, units, inboundExecutors);
|
||||||
},
|
},
|
||||||
() -> {
|
() -> {
|
||||||
if (shutdownExecutors)
|
if (shutdownExecutors)
|
||||||
|
|
@ -486,6 +492,30 @@ public final class MessagingService extends MessagingServiceMBeanImpl
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public void shutdownAbrubtly()
|
||||||
|
{
|
||||||
|
logger.debug("Shutting down abruptly");
|
||||||
|
if (isShuttingDown)
|
||||||
|
{
|
||||||
|
logger.info("Shutdown was already called");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
isShuttingDown = true;
|
||||||
|
logger.info("Waiting for messaging service to quiesce");
|
||||||
|
// We may need to schedule hints on the mutation stage, so it's erroneous to shut down the mutation stage first
|
||||||
|
assert !MUTATION.executor().isShutdown();
|
||||||
|
|
||||||
|
callbacks.shutdownNow(false);
|
||||||
|
inboundSockets.close();
|
||||||
|
for (OutboundConnections pool : channelManagers.values())
|
||||||
|
pool.close(false);
|
||||||
|
|
||||||
|
maybeFail(socketFactory::shutdownNow,
|
||||||
|
inboundSink::clear,
|
||||||
|
outboundSink::clear);
|
||||||
|
}
|
||||||
|
|
||||||
private void shutdownExecutors(long deadlineNanos) throws TimeoutException, InterruptedException
|
private void shutdownExecutors(long deadlineNanos) throws TimeoutException, InterruptedException
|
||||||
{
|
{
|
||||||
socketFactory.shutdownNow();
|
socketFactory.shutdownNow();
|
||||||
|
|
|
||||||
|
|
@ -73,7 +73,7 @@ final class SchemaKeyspace
|
||||||
|
|
||||||
private static final Logger logger = LoggerFactory.getLogger(SchemaKeyspace.class);
|
private static final Logger logger = LoggerFactory.getLogger(SchemaKeyspace.class);
|
||||||
|
|
||||||
private static final boolean FLUSH_SCHEMA_TABLES = Boolean.parseBoolean(System.getProperty("cassandra.test.flush_local_schema_changes", "true"));
|
private static final boolean FLUSH_SCHEMA_TABLES = CassandraRelevantProperties.FLUSH_LOCAL_SCHEMA_CHANGES.getBoolean();
|
||||||
private static final boolean IGNORE_CORRUPTED_SCHEMA_TABLES = Boolean.parseBoolean(System.getProperty("cassandra.ignore_corrupted_schema_tables", "false"));
|
private static final boolean IGNORE_CORRUPTED_SCHEMA_TABLES = Boolean.parseBoolean(System.getProperty("cassandra.ignore_corrupted_schema_tables", "false"));
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
||||||
|
|
@ -30,10 +30,12 @@ import java.security.ProtectionDomain;
|
||||||
import java.util.Collections;
|
import java.util.Collections;
|
||||||
import java.util.Enumeration;
|
import java.util.Enumeration;
|
||||||
|
|
||||||
import io.netty.util.concurrent.FastThreadLocal;
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
|
||||||
import org.apache.cassandra.utils.logging.LoggingSupportFactory;
|
import io.netty.util.concurrent.FastThreadLocal;
|
||||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||||
|
import org.apache.cassandra.utils.logging.LoggingSupportFactory;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Custom {@link SecurityManager} and {@link Policy} implementation that only performs access checks
|
* Custom {@link SecurityManager} and {@link Policy} implementation that only performs access checks
|
||||||
|
|
@ -46,6 +48,8 @@ import org.apache.cassandra.config.DatabaseDescriptor;
|
||||||
*/
|
*/
|
||||||
public final class ThreadAwareSecurityManager extends SecurityManager
|
public final class ThreadAwareSecurityManager extends SecurityManager
|
||||||
{
|
{
|
||||||
|
private static final Logger logger = LoggerFactory.getLogger(ThreadAwareSecurityManager.class);
|
||||||
|
|
||||||
public static final PermissionCollection noPermissions = new PermissionCollection()
|
public static final PermissionCollection noPermissions = new PermissionCollection()
|
||||||
{
|
{
|
||||||
public void add(Permission permission)
|
public void add(Permission permission)
|
||||||
|
|
@ -81,6 +85,14 @@ public final class ThreadAwareSecurityManager extends SecurityManager
|
||||||
{
|
{
|
||||||
if (installed)
|
if (installed)
|
||||||
return;
|
return;
|
||||||
|
|
||||||
|
// this line is needed - we need to make sure AccessControlException is loaded before we install this SM
|
||||||
|
// otherwise we may get into stackoverflow when javax.security is not allowed package, and ACE is tried to be
|
||||||
|
// loaded when it is going to be thrown from SM (class loader triggers SM to verify javax.security,
|
||||||
|
// it recognizes it as not allowed and attempts to throw it...)
|
||||||
|
//noinspection PlaceholderCountMatchesArgumentCount
|
||||||
|
logger.trace("Initialized thread aware security manager", AccessControlException.class.getName());
|
||||||
|
|
||||||
System.setSecurityManager(new ThreadAwareSecurityManager());
|
System.setSecurityManager(new ThreadAwareSecurityManager());
|
||||||
LoggingSupportFactory.getLoggingSupport().onStartup();
|
LoggingSupportFactory.getLoggingSupport().onStartup();
|
||||||
installed = true;
|
installed = true;
|
||||||
|
|
|
||||||
|
|
@ -48,6 +48,7 @@ import com.google.common.collect.Sets;
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
import org.slf4j.LoggerFactory;
|
import org.slf4j.LoggerFactory;
|
||||||
|
|
||||||
|
import org.apache.cassandra.config.CassandraRelevantProperties;
|
||||||
import org.apache.cassandra.db.ColumnFamilyStore;
|
import org.apache.cassandra.db.ColumnFamilyStore;
|
||||||
import org.apache.cassandra.db.Keyspace;
|
import org.apache.cassandra.db.Keyspace;
|
||||||
import org.apache.cassandra.dht.IPartitioner;
|
import org.apache.cassandra.dht.IPartitioner;
|
||||||
|
|
@ -163,6 +164,12 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
|
||||||
{
|
{
|
||||||
private INodeProvisionStrategy.Strategy nodeProvisionStrategy = INodeProvisionStrategy.Strategy.MultipleNetworkInterfaces;
|
private INodeProvisionStrategy.Strategy nodeProvisionStrategy = INodeProvisionStrategy.Strategy.MultipleNetworkInterfaces;
|
||||||
|
|
||||||
|
{
|
||||||
|
// those properties may be set for unit-test optimizations; those should not be used when running dtests
|
||||||
|
CassandraRelevantProperties.FLUSH_LOCAL_SCHEMA_CHANGES.reset();
|
||||||
|
CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN.reset();
|
||||||
|
}
|
||||||
|
|
||||||
public AbstractBuilder(Factory<I, C, B> factory)
|
public AbstractBuilder(Factory<I, C, B> factory)
|
||||||
{
|
{
|
||||||
super(factory);
|
super(factory);
|
||||||
|
|
@ -941,4 +948,3 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
|
||||||
.collect(Collectors.toSet());
|
.collect(Collectors.toSet());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -56,6 +56,7 @@ import com.datastax.driver.core.*;
|
||||||
import com.datastax.driver.core.DataType;
|
import com.datastax.driver.core.DataType;
|
||||||
import com.datastax.driver.core.ResultSet;
|
import com.datastax.driver.core.ResultSet;
|
||||||
|
|
||||||
|
import com.datastax.shaded.netty.channel.EventLoopGroup;
|
||||||
import org.apache.cassandra.SchemaLoader;
|
import org.apache.cassandra.SchemaLoader;
|
||||||
import org.apache.cassandra.ServerTestUtils;
|
import org.apache.cassandra.ServerTestUtils;
|
||||||
import org.apache.cassandra.concurrent.ScheduledExecutors;
|
import org.apache.cassandra.concurrent.ScheduledExecutors;
|
||||||
|
|
@ -134,6 +135,16 @@ public abstract class CQLTester
|
||||||
CREATE_INDEX_NAME_REGEX);
|
CREATE_INDEX_NAME_REGEX);
|
||||||
private static final Pattern CREATE_INDEX_PATTERN = Pattern.compile(CREATE_INDEX_REGEX, Pattern.CASE_INSENSITIVE);
|
private static final Pattern CREATE_INDEX_PATTERN = Pattern.compile(CREATE_INDEX_REGEX, Pattern.CASE_INSENSITIVE);
|
||||||
|
|
||||||
|
public static final NettyOptions IMMEDIATE_CONNECTION_SHUTDOWN_NETTY_OPTIONS = new NettyOptions()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public void onClusterClose(EventLoopGroup eventLoopGroup)
|
||||||
|
{
|
||||||
|
// shutdown driver connection immediatelly
|
||||||
|
eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.SECONDS).syncUninterruptibly();
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
/** Return the current server version if supported by the driver, else
|
/** Return the current server version if supported by the driver, else
|
||||||
* the latest that is supported.
|
* the latest that is supported.
|
||||||
*
|
*
|
||||||
|
|
@ -482,9 +493,10 @@ public abstract class CQLTester
|
||||||
.addContactPoints(nativeAddr)
|
.addContactPoints(nativeAddr)
|
||||||
.withClusterName("Test Cluster")
|
.withClusterName("Test Cluster")
|
||||||
.withPort(nativePort)
|
.withPort(nativePort)
|
||||||
.withSocketOptions(socketOptions);
|
.withSocketOptions(socketOptions)
|
||||||
|
.withNettyOptions(IMMEDIATE_CONNECTION_SHUTDOWN_NETTY_OPTIONS);
|
||||||
|
|
||||||
if (clusterConfigurator != null)
|
if (clusterConfigurator != null)
|
||||||
clusterConfigurator.accept(builder);
|
clusterConfigurator.accept(builder);
|
||||||
|
|
||||||
if (version.isBeta())
|
if (version.isBeta())
|
||||||
|
|
|
||||||
|
|
@ -36,7 +36,9 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
|
||||||
import org.apache.cassandra.metrics.CassandraMetricsRegistry;
|
import org.apache.cassandra.metrics.CassandraMetricsRegistry;
|
||||||
import org.apache.cassandra.metrics.TableMetrics;
|
import org.apache.cassandra.metrics.TableMetrics;
|
||||||
import org.apache.cassandra.schema.KeyspaceParams;
|
import org.apache.cassandra.schema.KeyspaceParams;
|
||||||
|
import org.apache.cassandra.schema.SchemaConstants;
|
||||||
import org.apache.cassandra.schema.TableMetadata;
|
import org.apache.cassandra.schema.TableMetadata;
|
||||||
|
import org.apache.cassandra.service.StorageService;
|
||||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||||
import org.apache.cassandra.utils.FBUtilities;
|
import org.apache.cassandra.utils.FBUtilities;
|
||||||
|
|
||||||
|
|
@ -46,12 +48,19 @@ import static org.junit.Assert.*;
|
||||||
public class ColumnFamilyMetricTest
|
public class ColumnFamilyMetricTest
|
||||||
{
|
{
|
||||||
@BeforeClass
|
@BeforeClass
|
||||||
public static void defineSchema()
|
public static void defineSchema() throws Exception
|
||||||
{
|
{
|
||||||
SchemaLoader.prepareServer();
|
SchemaLoader.prepareServer();
|
||||||
SchemaLoader.createKeyspace("Keyspace1",
|
SchemaLoader.createKeyspace("Keyspace1",
|
||||||
KeyspaceParams.simple(1),
|
KeyspaceParams.simple(1),
|
||||||
SchemaLoader.standardCFMD("Keyspace1", "Standard2"));
|
SchemaLoader.standardCFMD("Keyspace1", "Standard2"));
|
||||||
|
|
||||||
|
// we need this to properly initialize various static fields in the whole system
|
||||||
|
// since for the unit tests automatic schema flushing is disabled, the first flush may happen unexpectedly
|
||||||
|
// late - after the whole system is already running, and some static fields may remain uninitialized
|
||||||
|
// OTOH, late initialization of them may have creepy effects (for example NPEs in static initializers)
|
||||||
|
// disclaimer: this is not a proper way to fix that
|
||||||
|
StorageService.instance.forceKeyspaceFlush(SchemaConstants.SYSTEM_KEYSPACE_NAME);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
|
|
|
||||||
|
|
@ -123,6 +123,8 @@ public class NodeToolTPStatsTest extends CQLTester
|
||||||
|
|
||||||
createTable("CREATE TABLE %s (pk int, c int, PRIMARY KEY(pk))");
|
createTable("CREATE TABLE %s (pk int, c int, PRIMARY KEY(pk))");
|
||||||
execute("INSERT INTO %s (pk, c) VALUES (?, ?)", 1, 1);
|
execute("INSERT INTO %s (pk, c) VALUES (?, ?)", 1, 1);
|
||||||
|
flush();
|
||||||
|
|
||||||
tool = ToolRunner.invokeNodetool("tpstats");
|
tool = ToolRunner.invokeNodetool("tpstats");
|
||||||
assertTrue(tool.getCleanedStderr().isEmpty());
|
assertTrue(tool.getCleanedStderr().isEmpty());
|
||||||
assertEquals(0, tool.getExitCode());
|
assertEquals(0, tool.getExitCode());
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue