diff --git a/CHANGES.txt b/CHANGES.txt
index 71696fb626..4c23416766 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -106,6 +106,7 @@ Merged from 4.1:
* Streaming progress virtual table lock contention can trigger TCP_USER_TIMEOUT and fail streaming (CASSANDRA-18110)
* Fix perpetual load of denylist on read in cases where denylist can never be loaded (CASSANDRA-18116)
Merged from 4.0:
+ * Improve unit tests performance (CASSANDRA-17427)
* Connect to listen address when own broadcast address is requested (CASSANDRA-18200)
* Add safeguard so cleanup fails when node has pending ranges (CASSANDRA-16418)
* Fix legacy clustering serialization for paging with compact storage (CASSANDRA-17507)
diff --git a/build.xml b/build.xml
index 3c29f8c365..0fd3d7824d 100644
--- a/build.xml
+++ b/build.xml
@@ -111,12 +111,18 @@
-
+
+
+
+
+
+
+
@@ -137,6 +143,10 @@
+
+
+
+
+
@@ -1122,6 +1133,9 @@
+
+
+
@@ -1797,6 +1811,7 @@
+ Apache Cassandra ${eclipse.project.name}
diff --git a/ide/idea/workspace.xml b/ide/idea/workspace.xml
index a89921cc77..b3ec20bbe1 100644
--- a/ide/idea/workspace.xml
+++ b/ide/idea/workspace.xml
@@ -161,13 +161,19 @@
+
+
+
+
+
+
-
+
diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
index 4d30df2dbe..71dfd7f364 100644
--- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
+++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
@@ -339,6 +339,23 @@ public enum CassandraRelevantProperties
* The default used in SimpleSeedProvider is 20.
*/
SEED_COUNT_WARN_THRESHOLD("cassandra.seed_count_warn_threshold"),
+
+
+ /** 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)
@@ -382,6 +399,17 @@ public enum CassandraRelevantProperties
return defaultVal;
}
+ /**
+ * 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.
* @return system property String value if it exists, overrideDefaultValue otherwise.
@@ -584,4 +612,3 @@ public enum CassandraRelevantProperties
return System.getProperties().containsKey(key);
}
}
-
diff --git a/src/java/org/apache/cassandra/io/util/PathUtils.java b/src/java/org/apache/cassandra/io/util/PathUtils.java
index bb797d0712..4b3efdb559 100644
--- a/src/java/org/apache/cassandra/io/util/PathUtils.java
+++ b/src/java/org/apache/cassandra/io/util/PathUtils.java
@@ -24,6 +24,7 @@ import java.nio.file.attribute.*;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.*;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.annotation.Nullable;
@@ -44,6 +45,7 @@ import org.apache.cassandra.utils.NoSpamLogger;
import static java.nio.file.StandardOpenOption.*;
import static java.util.Collections.unmodifiableSet;
+import static org.apache.cassandra.config.CassandraRelevantProperties.USE_NIX_RECURSIVE_DELETE;
import static org.apache.cassandra.utils.Throwables.merge;
/**
@@ -71,7 +73,7 @@ public final class PathUtils
if (StorageService.instance.isDaemonSetupCompleted())
setDeletionListener(ignore -> {});
else
- logger.info("Deleting file during startup: {}", path);
+ logger.trace("Deleting file during startup: {}", path);
};
public static FileChannel newReadChannel(Path path) throws NoSuchFileException
@@ -310,6 +312,55 @@ public final class PathUtils
return accumulate;
}
+ /**
+ * Uses unix `rm -r` to delete a directory recursively.
+ * Note that, it will trigger {@link #onDeletion} listener only for the provided path and will not call it for any
+ * nested path. 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 ? "-rdf" : "-rd", path.toAbsolutePath().toString() };
+ try
+ {
+ if (!quietly && !Files.exists(path))
+ throw new NoSuchFileException(path.toString());
+
+ 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));
+ }
+
+ onDeletion.accept(path);
+ }
+ catch (IOException e)
+ {
+ throw propagateUnchecked(e, path, true);
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ throw new FSWriteError(e, path);
+ }
+ }
+
/**
* Deletes all files and subdirectories under "path".
* @param path file to be deleted
@@ -317,10 +368,16 @@ public final class PathUtils
*/
public static void deleteRecursive(Path path)
{
+ if (USE_NIX_RECURSIVE_DELETE.getBoolean() && path.getFileSystem() == FileSystems.getDefault())
+ {
+ deleteRecursiveUsingNixCommand(path, false);
+ return;
+ }
+
if (isDirectory(path))
forEach(path, PathUtils::deleteRecursive);
- // The directory is now empty so now it can be smoked
+ // The directory is now empty, so now it can be smoked
delete(path);
}
@@ -331,6 +388,12 @@ public final class PathUtils
*/
public static void deleteRecursive(Path path, RateLimiter rateLimiter)
{
+ if (USE_NIX_RECURSIVE_DELETE.getBoolean() && path.getFileSystem() == FileSystems.getDefault())
+ {
+ deleteRecursiveUsingNixCommand(path, false);
+ return;
+ }
+
deleteRecursive(path, rateLimiter, p -> deleteRecursive(p, rateLimiter));
}
diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java
index 650bc03e9e..a6a6fe60e4 100644
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@ -51,6 +51,7 @@ import org.apache.cassandra.utils.FBUtilities;
import static java.util.Collections.synchronizedList;
import static java.util.concurrent.TimeUnit.MINUTES;
import static org.apache.cassandra.concurrent.Stage.MUTATION;
+import static org.apache.cassandra.config.CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.Throwables.maybeFail;
@@ -523,15 +524,20 @@ public 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()
{
- 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)
{
+ logger.debug("Shutting down: timeout={}s, gracefully={}, shutdownExecutors={}", units.toSeconds(timeout), shutdownGracefully, shutdownExecutors);
if (isShuttingDown)
{
logger.info("Shutdown was already called");
@@ -555,7 +561,7 @@ public class MessagingService extends MessagingServiceMBeanImpl
() -> {
List inboundExecutors = new ArrayList<>();
inboundSockets.close(synchronizedList(inboundExecutors)::add).get();
- ExecutorUtils.awaitTermination(1L, TimeUnit.MINUTES, inboundExecutors);
+ ExecutorUtils.awaitTermination(timeout, units, inboundExecutors);
},
() -> {
if (shutdownExecutors)
@@ -587,6 +593,30 @@ public 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
{
socketFactory.shutdownNow();
diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
index 8059e51b76..ee33c00096 100644
--- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
+++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
@@ -76,7 +76,7 @@ public final class SchemaKeyspace
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"));
/**
diff --git a/src/java/org/apache/cassandra/security/ThreadAwareSecurityManager.java b/src/java/org/apache/cassandra/security/ThreadAwareSecurityManager.java
index 3ff6820714..8f86831cc3 100644
--- a/src/java/org/apache/cassandra/security/ThreadAwareSecurityManager.java
+++ b/src/java/org/apache/cassandra/security/ThreadAwareSecurityManager.java
@@ -30,10 +30,12 @@ import java.security.ProtectionDomain;
import java.util.Collections;
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.utils.logging.LoggingSupportFactory;
/**
* 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
{
+ private static final Logger logger = LoggerFactory.getLogger(ThreadAwareSecurityManager.class);
+
public static final PermissionCollection noPermissions = new PermissionCollection()
{
public void add(Permission permission)
@@ -81,6 +85,14 @@ public final class ThreadAwareSecurityManager extends SecurityManager
{
if (installed)
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());
LoggingSupportFactory.getLoggingSupport().onStartup();
installed = true;
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
index aad18243be..7d333726e5 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
@@ -60,6 +60,7 @@ import org.junit.Assume;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.IPartitioner;
@@ -184,6 +185,12 @@ public abstract class AbstractCluster implements ICluster factory)
{
super(factory);
@@ -1319,4 +1326,3 @@ public abstract class AbstractCluster implements ICluster