diff --git a/.circleci/config.yml b/.circleci/config.yml index 1a3f56af03..89c25cd267 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -171,8 +171,6 @@ jobs: # get all of our unit test filenames set -eo pipefail && circleci tests glob "$HOME/cassandra/test/unit/**/*.java" > /tmp/all_java_unit_tests.txt - # append distributed tests - set -eo pipefail && circleci tests glob "$HOME/cassandra/test/distributed/**/test/*.java" > /tmp/all_java_distributed_tests.txt # split up the unit tests into groups based on the number of containers we have set -eo pipefail && circleci tests split --split-by=timings --timings-type=filename --index=${CIRCLE_NODE_INDEX} --total=${CIRCLE_NODE_TOTAL} /tmp/all_java_unit_tests.txt > /tmp/java_tests_${CIRCLE_NODE_INDEX}.txt @@ -180,11 +178,6 @@ jobs: echo "** /tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt" cat /tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt - set -eo pipefail && circleci tests split --split-by=timings --timings-type=filename --index=${CIRCLE_NODE_INDEX} --total=${CIRCLE_NODE_TOTAL} /tmp/all_java_distributed_tests.txt > /tmp/java_dtests_${CIRCLE_NODE_INDEX}.txt - set +eo pipefail && cat /tmp/java_dtests_${CIRCLE_NODE_INDEX}.txt | cut -c 44-1000000 | grep "Test\.java$" > /tmp/java_dtests_${CIRCLE_NODE_INDEX}_final.txt - echo "** /tmp/java_dtests_${CIRCLE_NODE_INDEX}_final.txt" - cat /tmp/java_dtests_${CIRCLE_NODE_INDEX}_final.txt - - run: name: Run Unit Tests command: | @@ -198,8 +191,8 @@ jobs: cd /tmp/cassandra ant testclasslist -Dtest.classlistfile=/tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt -Dtest.classlistprefix=unit - if [ -s "/tmp/java_dtests_${CIRCLE_NODE_INDEX}_final.txt" ]; then - ant testclasslist -Dtest.classlistfile=/tmp/java_dtests_${CIRCLE_NODE_INDEX}_final.txt -Dtest.classlistprefix=distributed + if [ "${CIRCLE_NODE_INDEX}" -eq "1" ]; then + ant test-jvm-dtest-forking fi no_output_timeout: 15m - store_test_results: diff --git a/build.xml b/build.xml index 7bfd261383..c95217b081 100644 --- a/build.xml +++ b/build.xml @@ -65,6 +65,7 @@ + @@ -98,7 +99,7 @@ - + @@ -1169,15 +1170,6 @@ - - - - - - - - - - + @@ -1607,6 +1600,7 @@ + @@ -1618,7 +1612,7 @@ + testtag="@{testtag}" showoutput="false" > @@ -1764,6 +1758,46 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + shutdown(); // these methods are not for external use, but for simplicity we leave them public and on the normal IInstance interface void startup(ICluster cluster); diff --git a/test/distributed/org/apache/cassandra/distributed/api/IIsolatedExecutor.java b/test/distributed/org/apache/cassandra/distributed/api/IIsolatedExecutor.java index f9698b0965..6bb41d7969 100644 --- a/test/distributed/org/apache/cassandra/distributed/api/IIsolatedExecutor.java +++ b/test/distributed/org/apache/cassandra/distributed/api/IIsolatedExecutor.java @@ -20,7 +20,6 @@ package org.apache.cassandra.distributed.api; import java.io.Serializable; import java.util.concurrent.Callable; -import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.function.BiConsumer; import java.util.function.BiFunction; @@ -52,7 +51,7 @@ public interface IIsolatedExecutor } public interface SerializableTriFunction extends Serializable, TriFunction { } - void shutdown(); + Future shutdown(); /** * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index 14ee1eeb74..2e759f5974 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -30,6 +30,9 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; @@ -143,13 +146,14 @@ public abstract class AbstractCluster implements ICluster, } @Override - public synchronized void shutdown() + public synchronized Future shutdown() { if (isShutdown) throw new IllegalStateException(); isShutdown = true; - delegate.shutdown(); + Future future = delegate.shutdown(); delegate = null; + return future; } @Override @@ -238,12 +242,14 @@ public abstract class AbstractCluster implements ICluster, public void schemaChange(String query) { - try (SchemaChangeMonitor monitor = new SchemaChangeMonitor()) - { - // execute the schema change - coordinator(1).execute(query, ConsistencyLevel.ALL); - monitor.waitForAgreement(); - } + get(1).sync(() -> { + try (SchemaChangeMonitor monitor = new SchemaChangeMonitor()) + { + // execute the schema change + coordinator(1).execute(query, ConsistencyLevel.ALL); + monitor.waitForAgreement(); + } + }).run(); } /** @@ -375,13 +381,17 @@ public abstract class AbstractCluster implements ICluster, @Override public void close() { - parallelForEach(IInstance::shutdown, 1L, TimeUnit.MINUTES); + FBUtilities.waitOnFutures(instances.stream() + .map(IInstance::shutdown) + .collect(Collectors.toList()), + 1L, TimeUnit.MINUTES); + instances.clear(); + instanceMap.clear(); // Make sure to only delete directory when threads are stopped FileUtils.deleteRecursive(root); //withThreadLeakCheck(futures); - System.gc(); } // We do not want this check to run every time until we fix problems with tread stops diff --git a/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java b/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java index 68f957baff..27e2c04d0f 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java @@ -92,9 +92,9 @@ public abstract class DelegatingInvokableInstance implements IInvokableInstance } @Override - public void shutdown() + public Future shutdown() { - delegate().shutdown(); + return delegate().shutdown(); } @Override diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index 656f3f6f3c..e0e1dc9058 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -27,6 +27,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.function.BiConsumer; @@ -35,6 +37,7 @@ import java.util.function.Function; import org.slf4j.LoggerFactory; import ch.qos.logback.classic.LoggerContext; +import com.codahale.metrics.MetricFilter; import org.apache.cassandra.batchlog.BatchlogManager; import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.concurrent.SharedExecutorPool; @@ -68,6 +71,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.DataInputBuffer; import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.metrics.CassandraMetricsRegistry; import org.apache.cassandra.net.IMessageSink; import org.apache.cassandra.net.MessageDeliveryTask; import org.apache.cassandra.net.MessageIn; @@ -356,9 +360,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance } } - public void shutdown() + public Future shutdown() { - sync((ExecutorService executor) -> { + Future future = async((ExecutorService executor) -> { Throwable error = null; error = parallelRun(error, executor, Gossiper.instance::stop, @@ -385,11 +389,14 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance StageManager::shutdownAndWait, SharedExecutorPool.SHARED::shutdown ); + LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); loggerContext.stop(); - super.shutdown(); Throwables.maybeFail(error); - }).accept(isolatedExecutor); + }).apply(isolatedExecutor); + + return CompletableFuture.runAsync(ThrowingRunnable.toRunnable(future::get), isolatedExecutor) + .thenRun(super::shutdown); } private static Throwable parallelRun(Throwable accumulate, ExecutorService runOn, ThrowingRunnable ... runnables) @@ -424,9 +431,4 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance } return accumulate; } - - public static interface ThrowingRunnable - { - public void run() throws Throwable; - } } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java index 863164f051..d82c9e49ab 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java @@ -26,10 +26,13 @@ import java.io.ObjectOutputStream; import java.io.Serializable; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.net.URLClassLoader; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.Consumer; @@ -37,7 +40,6 @@ import java.util.function.Function; import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.distributed.api.IIsolatedExecutor; -import org.apache.cassandra.utils.Throwables; public class IsolatedExecutor implements IIsolatedExecutor { @@ -52,9 +54,12 @@ public class IsolatedExecutor implements IIsolatedExecutor this.deserializeOnInstance = lookupDeserializeOneObject(classLoader); } - public void shutdown() + public Future shutdown() { isolatedExecutor.shutdown(); + ThrowingRunnable.toRunnable(((URLClassLoader) classLoader)::close).run(); + return CompletableFuture.runAsync(ThrowingRunnable.toRunnable(() -> isolatedExecutor.awaitTermination(60, TimeUnit.SECONDS)), + Executors.newSingleThreadExecutor()); } public CallableNoExcept> async(CallableNoExcept call) { return () -> isolatedExecutor.submit(call); } @@ -162,4 +167,22 @@ public class IsolatedExecutor implements IIsolatedExecutor } } + public interface ThrowingRunnable + { + public void run() throws Throwable; + + public static Runnable toRunnable(ThrowingRunnable runnable) + { + return () -> { + try + { + runnable.run(); + } + catch (Throwable throwable) + { + throw new RuntimeException(throwable); + } + }; + } + } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java index e442603be7..e2c5303302 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java @@ -20,6 +20,8 @@ package org.apache.cassandra.distributed.test; import java.util.Arrays; +import com.google.common.collect.Iterators; +import org.junit.After; import org.junit.Assert; import org.junit.BeforeClass; @@ -27,6 +29,13 @@ import org.apache.cassandra.distributed.impl.AbstractCluster; public class DistributedTestBase { + @After + public void afterEach() + { + System.runFinalization(); + System.gc(); + } + public static String KEYSPACE = "distributed_test_keyspace"; @BeforeClass diff --git a/test/distributed/org/apache/cassandra/distributed/test/TestLocator.java b/test/distributed/org/apache/cassandra/distributed/test/TestLocator.java new file mode 100644 index 0000000000..a7ad400566 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/TestLocator.java @@ -0,0 +1,93 @@ +/* + * 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.distributed.test; + +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.io.PrintWriter; +import java.lang.reflect.Method; +import java.net.URL; +import java.util.ArrayList; +import java.util.Enumeration; +import java.util.List; + +import org.junit.Test; + +public class TestLocator +{ + private static final String defaultOutputFileName = "run-jvm-dtests"; + private static final String testPackage = "org.apache.cassandra.distributed.test"; + private static final String testCommandFormat = "ant testsome -Dtest.name=%s -Dtest.methods=%s"; + + public static void main(String[] args) throws Throwable + { + String outputFileName = defaultOutputFileName; + if (args.length == 1) + { + outputFileName = args[0]; + } + try (FileWriter fileWriter = new FileWriter(outputFileName); + PrintWriter printWriter = new PrintWriter(fileWriter)) + { + printWriter.println("#!/bin/bash"); + for (Class testClass : locateClasses(testPackage)) + { + for (Method method : testClass.getMethods()) + { + if (method.getAnnotation(Test.class) == null) + continue; + + printWriter.println(String.format(testCommandFormat, + testClass.getName(), + method.getName())); + } + } + } + } + + private static List locateClasses(String packageName) throws ClassNotFoundException, IOException + { + ClassLoader classLoader = TestLocator.class.getClassLoader(); + + Enumeration resources = classLoader.getResources(packageName.replace('.', '/')); + List classes = new ArrayList<>(); + while (resources.hasMoreElements()) + { + URL resource = resources.nextElement(); + loadClassesRecursively(new File(resource.getFile()), packageName, classes); + } + + return classes; + } + + + private static void loadClassesRecursively(File directory, String packageName, List classes) throws ClassNotFoundException + { + for (File file : directory.listFiles()) + { + if (file.isDirectory()) + loadClassesRecursively(file, packageName + "." + file.getName(), classes); + else if (file.getName().endsWith(".class")) + { + classes.add(Class.forName(packageName + '.' + file.getName().substring(0, file.getName().length() - 6))); + } + } + } +} \ No newline at end of file