diff --git a/.circleci/config.yml b/.circleci/config.yml index 791c49c1a9..1a3f56af03 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -172,7 +172,7 @@ 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/**/*.java" > /tmp/all_java_distributed_tests.txt + 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 diff --git a/CHANGES.txt b/CHANGES.txt index a7e35ea939..e3b23eea46 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,7 +1,3 @@ -3.0.19 - * Catch CorruptSSTableExceptions and FSErrors in ALAExecutorService (CASSANDRA-14993) - - 3.0.18 * Severe concurrency issues in STCS,DTCS,TWCS,TMD.Topology,TypeParser * Add a script to make running the cqlsh tests in cassandra repo easier (CASSANDRA-14951) diff --git a/build.xml b/build.xml index 610176beb6..d8dba625bd 100644 --- a/build.xml +++ b/build.xml @@ -880,6 +880,16 @@ + + + + + + + + + + @@ -1159,6 +1169,15 @@ + + + + + + + + + - + diff --git a/test/distributed/org/apache/cassandra/distributed/Cluster.java b/test/distributed/org/apache/cassandra/distributed/Cluster.java new file mode 100644 index 0000000000..c7f7675cb8 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/Cluster.java @@ -0,0 +1,57 @@ +/* + * 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; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.util.List; + +import org.apache.cassandra.distributed.api.ICluster; +import org.apache.cassandra.distributed.impl.AbstractCluster; +import org.apache.cassandra.distributed.impl.IInvokableInstance; +import org.apache.cassandra.distributed.impl.InstanceConfig; +import org.apache.cassandra.distributed.impl.Versions; + +/** + * A simple cluster supporting only the 'current' Cassandra version, offering easy access to the convenience methods + * of IInvokableInstance on each node. + */ +public class Cluster extends AbstractCluster implements ICluster, AutoCloseable +{ + private Cluster(File root, Versions.Version version, List configs, ClassLoader sharedClassLoader) + { + super(root, version, configs, sharedClassLoader); + } + + protected IInvokableInstance newInstanceWrapper(Versions.Version version, InstanceConfig config) + { + return new Wrapper(version, config); + } + + public static Cluster create(int nodeCount) throws Throwable + { + return create(nodeCount, Cluster::new); + } + public static Cluster create(int nodeCount, File root) + { + return create(nodeCount, Versions.CURRENT, root, Cluster::new); + } +} + diff --git a/test/distributed/org/apache/cassandra/distributed/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/Coordinator.java deleted file mode 100644 index deadb2f694..0000000000 --- a/test/distributed/org/apache/cassandra/distributed/Coordinator.java +++ /dev/null @@ -1,78 +0,0 @@ -/* - * 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; - - -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.List; - -import org.apache.cassandra.cql3.CQLStatement; -import org.apache.cassandra.cql3.QueryOptions; -import org.apache.cassandra.cql3.QueryProcessor; -import org.apache.cassandra.db.ConsistencyLevel; -import org.apache.cassandra.service.ClientState; -import org.apache.cassandra.service.QueryState; -import org.apache.cassandra.transport.Server; -import org.apache.cassandra.transport.messages.ResultMessage; -import org.apache.cassandra.utils.ByteBufferUtil; - - -public class Coordinator -{ - final Instance instance; - - public Coordinator(Instance instance) - { - this.instance = instance; - } - - private static Object[][] coordinatorExecute(String query, int consistencyLevel, Object[] bindings) - { - CQLStatement prepared = QueryProcessor.getStatement(query, ClientState.forInternalCalls()).statement; - List boundValues = new ArrayList<>(); - for (Object binding : bindings) - { - boundValues.add(ByteBufferUtil.objectToBytes(binding)); - } - - ResultMessage res = prepared.execute(QueryState.forInternalCalls(), - QueryOptions.create(ConsistencyLevel.fromCode(consistencyLevel), - boundValues, - false, - 10, - null, - null, - Server.VERSION_4)); - - if (res != null && res.kind == ResultMessage.Kind.ROWS) - { - return RowUtil.toObjects((ResultMessage.Rows) res); - } - else - { - return new Object[][]{}; - } - } - - public Object[][] execute(String query, ConsistencyLevel consistencyLevel, Object... boundValues) - { - return instance.appliesOnInstance(Coordinator::coordinatorExecute).apply(query, consistencyLevel.code, boundValues); - } -} diff --git a/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java b/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java deleted file mode 100644 index 9958fd3960..0000000000 --- a/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java +++ /dev/null @@ -1,109 +0,0 @@ -/* - * 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; - -import com.google.common.base.Predicate; -import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.utils.Pair; - -import java.net.URL; -import java.net.URLClassLoader; -import java.util.HashSet; -import java.util.Set; -import java.util.function.IntFunction; - -public class InstanceClassLoader extends URLClassLoader -{ - // Classes that have to be shared between instances, for configuration or returning values - private final static Class[] commonClasses = new Class[] - { - Pair.class, - InstanceConfig.class, - Message.class, - InetAddressAndPort.class, - InvokableInstance.SerializableBiConsumer.class, - InvokableInstance.SerializableBiFunction.class, - InvokableInstance.SerializableCallable.class, - InvokableInstance.SerializableConsumer.class, - InvokableInstance.SerializableFunction.class, - InvokableInstance.SerializableRunnable.class, - InvokableInstance.SerializableTriFunction.class, - InvokableInstance.InstanceFunction.class - }; - - private final int id; // for debug purposes - private final ClassLoader commonClassLoader; - private final Predicate isCommonClassName; - - InstanceClassLoader(int id, URL[] urls, Predicate isCommonClassName, ClassLoader commonClassLoader) - { - super(urls, null); - this.id = id; - this.commonClassLoader = commonClassLoader; - this.isCommonClassName = isCommonClassName; - } - - @Override - public Class loadClass(String name) throws ClassNotFoundException - { - // Do not share: - // * yaml, which is a rare exception because it does mess with loading org.cassandra...Config class instances - // * most of the rest of Cassandra classes (unless they were explicitly shared) g - if (name.startsWith("org.slf4j") || - name.startsWith("ch.qos.logback") || - name.startsWith("org.yaml") || - (name.startsWith("org.apache.cassandra") && !isCommonClassName.apply(name))) - return loadClassInternal(name); - - return commonClassLoader.loadClass(name); - } - - Class loadClassInternal(String name) throws ClassNotFoundException - { - synchronized (getClassLoadingLock(name)) - { - // First, check if the class has already been loaded - Class c = findLoadedClass(name); - - if (c == null) - c = findClass(name); - - return c; - } - } - - public static IntFunction createFactory(URLClassLoader contextClassLoader) - { - Set commonClassNames = new HashSet<>(); - for (Class k : commonClasses) - commonClassNames.add(k.getName()); - - URL[] urls = contextClassLoader.getURLs(); - return id -> new InstanceClassLoader(id, urls, commonClassNames::contains, contextClassLoader); - } - - /** - * @return true iff this class was loaded by an InstanceClassLoader, and as such is used by a dtest node - */ - public static boolean wasLoadedByAnInstanceClassLoader(Class clazz) - { - return clazz.getClassLoader().getClass().getName().equals(InstanceClassLoader.class.getName()); - } - -} diff --git a/test/distributed/org/apache/cassandra/distributed/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/InstanceConfig.java deleted file mode 100644 index 49c2e1f624..0000000000 --- a/test/distributed/org/apache/cassandra/distributed/InstanceConfig.java +++ /dev/null @@ -1,87 +0,0 @@ -/* - * 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; - -import java.io.File; -import java.io.Serializable; -import java.util.UUID; - -public class InstanceConfig implements Serializable -{ - public final int num; - public final UUID hostId =java.util.UUID.randomUUID(); - public final String partitioner = "org.apache.cassandra.dht.Murmur3Partitioner"; - public final String broadcast_address; - public final String listen_address; - public final String broadcast_rpc_address; - public final String rpc_address; - public final String saved_caches_directory; - public final String[] data_file_directories; - public final String commitlog_directory; - public final String hints_directory; - public final String cdc_directory; - public final int concurrent_writes = 2; - public final int concurrent_counter_writes = 2; - public final int concurrent_materialized_view_writes = 2; - public final int concurrent_reads = 2; - public final int memtable_flush_writers = 1; - public final int concurrent_compactors = 1; - public final int memtable_heap_space_in_mb = 10; - public final String initial_token; - - private InstanceConfig(int num, - String broadcast_address, - String listen_address, - String broadcast_rpc_address, - String rpc_address, - String saved_caches_directory, - String[] data_file_directories, - String commitlog_directory, - String hints_directory, - String cdc_directory, - String initial_token) - { - this.num = num; - this.broadcast_address = broadcast_address; - this.listen_address = listen_address; - this.broadcast_rpc_address = broadcast_rpc_address; - this.rpc_address = rpc_address; - this.saved_caches_directory = saved_caches_directory; - this.data_file_directories = data_file_directories; - this.commitlog_directory = commitlog_directory; - this.hints_directory = hints_directory; - this.cdc_directory = cdc_directory; - this.initial_token = initial_token; - } - - public static InstanceConfig generate(int nodeNum, File root, String token) - { - return new InstanceConfig(nodeNum, - "127.0.0." + nodeNum, - "127.0.0." + nodeNum, - "127.0.0." + nodeNum, - "127.0.0." + nodeNum, - String.format("%s/node%d/saved_caches", root, nodeNum), - new String[] { String.format("%s/node%d/data", root, nodeNum) }, - String.format("%s/node%d/commitlog", root, nodeNum), - String.format("%s/node%d/hints", root, nodeNum), - String.format("%s/node%d/cdc", root, nodeNum), - token); - } -} diff --git a/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java b/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java deleted file mode 100644 index 9fb543d288..0000000000 --- a/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java +++ /dev/null @@ -1,213 +0,0 @@ -/* - * 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; - -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.Closeable; -import java.io.IOException; -import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; -import java.io.Serializable; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.function.BiConsumer; -import java.util.function.BiFunction; -import java.util.function.Consumer; -import java.util.function.Function; - -import org.apache.cassandra.concurrent.NamedThreadFactory; -import org.apache.cassandra.utils.Throwables; - -public abstract class InvokableInstance -{ - protected final ExecutorService isolatedExecutor; - private final ClassLoader classLoader; - private final Method deserializeOnInstance; - - public InvokableInstance(String name, ClassLoader classLoader) - { - this.isolatedExecutor = Executors.newCachedThreadPool(new NamedThreadFactory(name, Thread.NORM_PRIORITY, classLoader, new ThreadGroup(name))); - this.classLoader = classLoader; - try - { - this.deserializeOnInstance = classLoader.loadClass(InvokableInstance.class.getName()).getDeclaredMethod("deserializeOneObject", byte[].class); - } - catch (ClassNotFoundException | NoSuchMethodException e) - { - throw new RuntimeException(e); - } - } - - public interface CallableNoExcept extends Callable { public T call(); } - public interface SerializableCallable extends CallableNoExcept, Serializable { } - public CallableNoExcept callsOnInstance(SerializableCallable call) { return invokesOnExecutor((SerializableCallable) transferOneObject(call), isolatedExecutor); } - public T callOnInstance(SerializableCallable call) { return callsOnInstance(call).call(); } - - public interface SerializableRunnable extends Runnable, Serializable {} - public Runnable runsOnInstance(SerializableRunnable run) { return invokesOnExecutor((SerializableRunnable) transferOneObject(run), isolatedExecutor); } - public void runOnInstance(SerializableRunnable run) { runsOnInstance(run).run(); } - - public interface SerializableConsumer extends Consumer, Serializable {} - public Consumer acceptsOnInstance(SerializableConsumer consumer) { return invokesOnExecutor((SerializableConsumer) transferOneObject(consumer), isolatedExecutor); } - - public interface SerializableBiConsumer extends BiConsumer, Serializable {} - public BiConsumer acceptsOnInstance(SerializableBiConsumer consumer) { return invokesOnExecutor((SerializableBiConsumer) transferOneObject(consumer), isolatedExecutor); } - - public interface SerializableFunction extends Function, Serializable {} - public Function appliesOnInstance(SerializableFunction f) { return invokesOnExecutor((SerializableFunction) transferOneObject(f), isolatedExecutor); } - - public interface SerializableBiFunction extends BiFunction, Serializable {} - public BiFunction appliesOnInstance(SerializableBiFunction f) { return invokesOnExecutor((SerializableBiFunction) transferOneObject(f), isolatedExecutor); } - - public interface TriFunction - { - O apply(I1 i1, I2 i2, I3 i3); - } - public interface SerializableTriFunction extends Serializable, TriFunction { } - - public TriFunction appliesOnInstance(SerializableTriFunction f) { return invokesOnExecutor((SerializableTriFunction) transferOneObject(f), isolatedExecutor); } - - public interface InstanceFunction extends SerializableBiFunction {} - - // E must be a functional interface, and lambda must be implemented by a lambda function - public E invokesOnInstance(E lambda) - { - return (E) transferOneObject(lambda); - } - - public Object transferOneObject(Object object) - { - byte[] bytes = serializeOneObject(object); - try - { - Object onInstance = deserializeOnInstance.invoke(null, bytes); - if (onInstance.getClass().getClassLoader() != classLoader) - throw new IllegalStateException(onInstance + " seemingly from wrong class loader: " + onInstance.getClass().getClassLoader() + ", but expected " + classLoader); - - return onInstance; - } - catch (IllegalAccessException | InvocationTargetException e) - { - throw new RuntimeException(e); - } - } - - private byte[] serializeOneObject(Object object) - { - try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); - ObjectOutputStream oos = new ObjectOutputStream(baos)) - { - oos.writeObject(object); - oos.close(); - return baos.toByteArray(); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - } - - @SuppressWarnings("unused") // called through method invocation - public static Object deserializeOneObject(byte[] bytes) - { - try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes); - ObjectInputStream ois = new ObjectInputStream(bais);) - { - return ois.readObject(); - } - catch (IOException | ClassNotFoundException e) - { - throw new RuntimeException(e); - } - } - - private static CallableNoExcept invokesOnExecutor(SerializableCallable callable, ExecutorService invokeOn) - { - return () -> { - try - { - return invokeOn.submit(callable).get(); - } - catch (InterruptedException e) - { - throw new RuntimeException(e); - } - catch (ExecutionException e) - { - Throwables.maybeFail(e.getCause()); - throw new AssertionError(); - } - }; - } - - private static Runnable invokesOnExecutor(SerializableRunnable runnable, ExecutorService invokeOn) - { - return () -> { - try - { - invokeOn.submit(runnable).get(); - } - catch (InterruptedException e) - { - throw new RuntimeException(e); - } - catch (ExecutionException e) - { - Throwables.maybeFail(e.getCause()); - throw new AssertionError(); - } - }; - } - - private static Consumer invokesOnExecutor(SerializableConsumer consumer, ExecutorService invokeOn) - { - return (a) -> invokesOnExecutor(() -> consumer.accept(a), invokeOn).run(); - } - - private static BiConsumer invokesOnExecutor(SerializableBiConsumer consumer, ExecutorService invokeOn) - { - return (a, b) -> invokesOnExecutor(() -> consumer.accept(a, b), invokeOn).run(); - } - - private static Function invokesOnExecutor(SerializableFunction f, ExecutorService invokeOn) - { - return (a) -> invokesOnExecutor(() -> f.apply(a), invokeOn).call(); - } - - private static BiFunction invokesOnExecutor(SerializableBiFunction f, ExecutorService invokeOn) - { - return (a, b) -> invokesOnExecutor(() -> f.apply(a, b), invokeOn).call(); - } - - private static SerializableTriFunction invokesOnExecutor(SerializableTriFunction f, ExecutorService invokeOn) - { - return (a, b, c) -> invokesOnExecutor(() -> f.apply(a, b, c), invokeOn).call(); - } - - void shutdown() - { - isolatedExecutor.shutdownNow(); - } - -} diff --git a/test/distributed/org/apache/cassandra/distributed/TestCluster.java b/test/distributed/org/apache/cassandra/distributed/TestCluster.java deleted file mode 100644 index 245030641d..0000000000 --- a/test/distributed/org/apache/cassandra/distributed/TestCluster.java +++ /dev/null @@ -1,280 +0,0 @@ -/* - * 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; - -import java.io.File; -import java.io.IOException; -import java.net.URLClassLoader; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.LockSupport; -import java.util.function.IntFunction; -import java.util.stream.Collectors; -import java.util.stream.Stream; - -import com.google.common.collect.Sets; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.cassandra.concurrent.NamedThreadFactory; -import org.apache.cassandra.db.ColumnFamilyStore; -import org.apache.cassandra.db.ConsistencyLevel; -import org.apache.cassandra.db.Keyspace; -import org.apache.cassandra.io.util.FileUtils; -import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.utils.FBUtilities; - -/** - * TestCluster creates, initializes and manages Cassandra instances ({@link Instance}. - * - * All instances created under the same cluster will have a shared ClassLoader that'll preload - * common classes required for configuration and communication (byte buffers, primitives, config - * objects etc). Shared classes are listed in {@link InstanceClassLoader#commonClasses}. - * - * Each instance has its own class loader that will load logging, yaml libraries and all non-shared - * Cassandra package classes. The rule of thumb is that we'd like to have all Cassandra-specific things - * (unless explitily shared through the common classloader) on a per-classloader basis in order to - * allow creating more than one instance of DatabaseDescriptor and other Cassandra singletones. - * - * All actions (reading, writing, schema changes, etc) are executed by serializing lambda/runnables, - * transferring them to instance-specific classloaders, deserializing and running them there. Most of - * the things can be simply captured in closure or passed through `apply` method of the wrapped serializable - * function/callable. You can use {@link InvokableInstance#{applies|runs|consumes}OnInstance} for executing - * code on specific instance. - * - * Each instance has its own logger. Each instance log line will contain INSTANCE{instance_id}. - * - * As of today, messaging is faked by hooking into MessagingService, so we're not using usual Cassandra - * handlers for internode to have more control over it. Messaging is wired by passing verbs manually. - * coordinator-handling code and hooks to the callbacks can be found in {@link Coordinator}. - */ -public class TestCluster implements AutoCloseable -{ - // WARNING: we have this logger not (necessarily) for logging, but - // to ensure we have instantiated the main classloader's LoggerFactory (and any LogbackStatusListener) - // before we instantiate any for a new instance - private static final Logger logger = LoggerFactory.getLogger(TestCluster.class); - - private final File root; - private final List instances; - private final Coordinator coordinator; - private final Map instanceMap; - private final MessageFilters filters; - - private TestCluster(File root, List instances) - { - this.root = root; - this.instances = instances; - this.instanceMap = new HashMap<>(); - this.coordinator = new Coordinator(instances.get(0)); - this.filters = new MessageFilters(this); - } - - void launch() - { - FBUtilities.waitOnFutures(instances.stream() - .map(i -> i.isolatedExecutor.submit(() -> i.launch(this))) - .collect(Collectors.toList()) - ); - for (Instance instance : instances) - instanceMap.put(instance.getBroadcastAddress(), instance); - } - - public int size() - { - return instances.size(); - } - - public Coordinator coordinator() - { - return coordinator; - } - - /** - * WARNING: we index from 1 here, for consistency with inet address! - */ - public Instance get(int idx) - { - return instances.get(idx - 1); - } - - public Stream stream() { return instances.stream(); } - - public Instance get(InetAddressAndPort addr) - { - return instanceMap.get(addr); - } - - MessageFilters filters() - { - return filters; - } - - MessageFilters.Builder verbs(MessagingService.Verb ... verbs) - { - return filters.verbs(verbs); - } - - public void disableAutoCompaction(String keyspace) - { - for (Instance instance : instances) - { - instance.runOnInstance(() -> { - for (ColumnFamilyStore cs : Keyspace.open(keyspace).getColumnFamilyStores()) - cs.disableAutoCompaction(); - }); - } - } - - public void schemaChange(String query) - { - try (SchemaChangeMonitor monitor = new SchemaChangeMonitor()) - { - // execute the schema change - coordinator().execute(query, ConsistencyLevel.ALL); - monitor.waitForAgreement(); - } - } - - /** - * Will wait for a schema change AND agreement that occurs after it is created - * (and precedes the invocation to waitForAgreement) - * - * Works by simply checking if all UUIDs agree after any schema version change event, - * so long as the waitForAgreement method has been entered (indicating the change has - * taken place on the coordinator) - * - * This could perhaps be made a little more robust, but this should more than suffice. - */ - public class SchemaChangeMonitor implements AutoCloseable - { - public SchemaChangeMonitor() {} - - @Override - public void close() { } - - public void waitForAgreement() - { - long start = System.nanoTime(); - while (1 != instances.stream().map(Instance::getSchemaVersion).distinct().count()) - { - if (System.nanoTime() - start > TimeUnit.MINUTES.toNanos(1L)) - throw new IllegalStateException("Schema agreement not reached"); - LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1L)); - } - } - } - - public void schemaChange(String statement, int instance) - { - get(instance).schemaChange(statement); - } - - public static TestCluster create(int nodeCount) throws Throwable - { - return create(nodeCount, Files.createTempDirectory("dtests").toFile()); - } - - public static TestCluster create(int nodeCount, File root) - { - root.mkdirs(); - setupLogging(root); - - IntFunction classLoaderFactory = InstanceClassLoader.createFactory( - (URLClassLoader) Thread.currentThread().getContextClassLoader()); - List instances = new ArrayList<>(); - long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount); - for (int i = 0 ; i < nodeCount ; ++i) - { - InstanceConfig instanceConfig = InstanceConfig.generate(i + 1, root, String.valueOf(token)); - instances.add(new Instance(instanceConfig, classLoaderFactory.apply(i + 1))); - token += increment; - } - - TestCluster cluster = new TestCluster(root, instances); - cluster.launch(); - return cluster; - } - - private static void setupLogging(File root) - { - try - { - String testConfPath = "test/conf/logback-dtest.xml"; - Path logConfPath = Paths.get(root.getPath(), "/logback-dtest.xml"); - if (!logConfPath.toFile().exists()) - { - Files.copy(new File(testConfPath).toPath(), - logConfPath); - } - System.setProperty("logback.configurationFile", "file://" + logConfPath); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - } - - @Override - public void close() - { - List> futures = instances.stream() - .map(i -> i.isolatedExecutor.submit(i::shutdown)) - .collect(Collectors.toList()); - - // Make sure to only delete directory when threads are stopped - FBUtilities.waitOnFutures(futures, 60, TimeUnit.SECONDS); - FileUtils.deleteRecursive(root); - - //withThreadLeakCheck(futures); - System.gc(); - } - - // We do not want this check to run every time until we fix problems with tread stops - private void withThreadLeakCheck(List> futures) - { - FBUtilities.waitOnFutures(futures); - - Set threadSet = Thread.getAllStackTraces().keySet(); - threadSet = Sets.difference(threadSet, Collections.singletonMap(Thread.currentThread(), null).keySet()); - if (!threadSet.isEmpty()) - { - for (Thread thread : threadSet) - { - System.out.println(thread); - System.out.println(Arrays.toString(thread.getStackTrace())); - } - throw new RuntimeException(String.format("Not all threads have shut down. %d threads are still running: %s", threadSet.size(), threadSet)); - } - } - -} - diff --git a/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java b/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java new file mode 100644 index 0000000000..0c8e63ae91 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java @@ -0,0 +1,70 @@ +/* + * 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; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.util.List; + +import org.apache.cassandra.distributed.api.ICluster; +import org.apache.cassandra.distributed.impl.AbstractCluster; +import org.apache.cassandra.distributed.impl.IUpgradeableInstance; +import org.apache.cassandra.distributed.impl.InstanceConfig; +import org.apache.cassandra.distributed.impl.Versions; + +/** + * A multi-version cluster, offering only the cross-version API + * + * TODO: we could perhaps offer some convenience methods for nodes we know to be of the 'current' version, + * to permit upgrade tests to perform cluster operations without updating the cross-version API, + * so long as one node is up-to-date. + */ +public class UpgradeableCluster extends AbstractCluster implements ICluster, AutoCloseable +{ + private UpgradeableCluster(File root, Versions.Version version, List configs, ClassLoader sharedClassLoader) + { + super(root, version, configs, sharedClassLoader); + } + + protected IUpgradeableInstance newInstanceWrapper(Versions.Version version, InstanceConfig config) + { + return new Wrapper(version, config); + } + + public static UpgradeableCluster create(int nodeCount) throws Throwable + { + return create(nodeCount, UpgradeableCluster::new); + } + public static UpgradeableCluster create(int nodeCount, File root) + { + return create(nodeCount, Versions.CURRENT, root, UpgradeableCluster::new); + } + + public static UpgradeableCluster create(int nodeCount, Versions.Version version) throws IOException + { + return create(nodeCount, version, Files.createTempDirectory("dtests").toFile(), UpgradeableCluster::new); + } + public static UpgradeableCluster create(int nodeCount, Versions.Version version, File root) + { + return create(nodeCount, version, root, UpgradeableCluster::new); + } + +} + diff --git a/test/distributed/org/apache/cassandra/distributed/api/ICluster.java b/test/distributed/org/apache/cassandra/distributed/api/ICluster.java new file mode 100644 index 0000000000..91da61ed34 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/api/ICluster.java @@ -0,0 +1,34 @@ +/* + * 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.api; + +import org.apache.cassandra.locator.InetAddressAndPort; + +import java.util.stream.Stream; + +public interface ICluster +{ + + IInstance get(int i); + IInstance get(InetAddressAndPort endpoint); + int size(); + Stream stream(); + IMessageFilters filters(); + +} diff --git a/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java b/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java new file mode 100644 index 0000000000..c7bdb36d30 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java @@ -0,0 +1,27 @@ +/* + * 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.api; + +// The cross-version API requires that a Coordinator can be constructed without any constructor arguments +public interface ICoordinator +{ + // a bit hacky, but ConsistencyLevel draws in too many dependent classes, so we cannot have a cross-version + // method signature that accepts ConsistencyLevel directly. So we just accept an Enum and cast. + Object[][] execute(String query, Enum consistencyLevel, Object... boundValues); +} diff --git a/test/distributed/org/apache/cassandra/distributed/api/IInstance.java b/test/distributed/org/apache/cassandra/distributed/api/IInstance.java new file mode 100644 index 0000000000..9cc022655d --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/api/IInstance.java @@ -0,0 +1,44 @@ +/* + * 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.api; + +import org.apache.cassandra.locator.InetAddressAndPort; + +import java.util.UUID; + +// The cross-version API requires that an Instance has a constructor signature of (IInstanceConfig, ClassLoader) +public interface IInstance extends IIsolatedExecutor +{ + ICoordinator coordinator(); + IListen listen(); + + void schemaChangeInternal(String query); + public Object[][] executeInternal(String query, Object... args); + + IInstanceConfig config(); + public InetAddressAndPort broadcastAddressAndPort(); + UUID schemaVersion(); + + void startup(); + void 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); + void receiveMessage(IMessage message); +} diff --git a/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java new file mode 100644 index 0000000000..6741b3fdd5 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java @@ -0,0 +1,41 @@ +/* + * 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.api; + +import org.apache.cassandra.locator.InetAddressAndPort; + +import java.util.UUID; + +public interface IInstanceConfig +{ + int num(); + UUID hostId(); + InetAddressAndPort broadcastAddressAndPort(); + + /** + * write the specified parameters to the Config object; we do not specify Config as the type to support a Config + * from any ClassLoader; the implementation must not directly access any fields of the Object, or cast it, but + * must use the reflection API to modify the state + */ + void propagate(Object writeToConfig); + + Object get(String fieldName); + String getString(String fieldName); + int getInt(String fieldName); +} diff --git a/test/distributed/org/apache/cassandra/distributed/api/IIsolatedExecutor.java b/test/distributed/org/apache/cassandra/distributed/api/IIsolatedExecutor.java new file mode 100644 index 0000000000..f9698b0965 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/api/IIsolatedExecutor.java @@ -0,0 +1,127 @@ +/* + * 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.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; +import java.util.function.Consumer; +import java.util.function.Function; + +/** + * Represents a clean way to handoff evaluation of some work to an executor associated + * with a node's lifetime. + * + * There is no transfer of execution to the parallel class hierarchy. + * + * Classes, such as Instance, that are themselves instantiated on the correct ClassLoader, utilise this class + * to ensure the lifetime of any thread evaluating one of its method invocations matches the lifetime of the class itself. + * Since they are instantiated on the correct ClassLoader, sharing only the interface, there is no serialization necessary. + */ +public interface IIsolatedExecutor +{ + public interface CallableNoExcept extends Callable { public O call(); } + public interface SerializableCallable extends CallableNoExcept, Serializable { } + public interface SerializableRunnable extends Runnable, Serializable {} + public interface SerializableConsumer extends Consumer, Serializable {} + public interface SerializableBiConsumer extends BiConsumer, Serializable {} + public interface SerializableFunction extends Function, Serializable {} + public interface SerializableBiFunction extends BiFunction, Serializable {} + public interface TriFunction + { + O apply(I1 i1, I2 i2, I3 i3); + } + public interface SerializableTriFunction extends Serializable, TriFunction { } + + void shutdown(); + + /** + * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result + */ + CallableNoExcept> async(CallableNoExcept call); + + /** + * Convert the execution to one performed synchronously on the IsolatedExecutor + */ + CallableNoExcept sync(CallableNoExcept call); + + /** + * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result + */ + CallableNoExcept> async(Runnable run); + + /** + * Convert the execution to one performed synchronously on the IsolatedExecutor + */ + Runnable sync(Runnable run); + + /** + * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result + */ + Function> async(Consumer consumer); + + /** + * Convert the execution to one performed synchronously on the IsolatedExecutor + */ + Consumer sync(Consumer consumer); + + /** + * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result + */ + BiFunction> async(BiConsumer consumer); + + /** + * Convert the execution to one performed synchronously on the IsolatedExecutor + */ + BiConsumer sync(BiConsumer consumer); + + /** + * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result + */ + Function> async(Function f); + + /** + * Convert the execution to one performed synchronously on the IsolatedExecutor + */ + Function sync(Function f); + + /** + * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result + */ + BiFunction> async(BiFunction f); + + /** + * Convert the execution to one performed synchronously on the IsolatedExecutor + */ + BiFunction sync(BiFunction f); + + /** + * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result + */ + TriFunction> async(TriFunction f); + + /** + * Convert the execution to one performed synchronously on the IsolatedExecutor + */ + TriFunction sync(TriFunction f); + +} diff --git a/test/distributed/org/apache/cassandra/distributed/api/IListen.java b/test/distributed/org/apache/cassandra/distributed/api/IListen.java new file mode 100644 index 0000000000..d3a80da468 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/api/IListen.java @@ -0,0 +1,26 @@ +/* + * 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.api; + +public interface IListen +{ + public interface Cancel { void cancel(); } + + Cancel schema(Runnable onChange); +} diff --git a/test/distributed/org/apache/cassandra/distributed/api/IMessage.java b/test/distributed/org/apache/cassandra/distributed/api/IMessage.java new file mode 100644 index 0000000000..1e537ed1ee --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/api/IMessage.java @@ -0,0 +1,33 @@ +/* + * 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.api; + +import org.apache.cassandra.locator.InetAddressAndPort; + +/** + * A cross-version interface for delivering internode messages via message sinks + */ +public interface IMessage +{ + int verb(); + byte[] bytes(); + int id(); + int version(); + InetAddressAndPort from(); +} diff --git a/test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java b/test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java similarity index 59% rename from test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java rename to test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java index 1ff88eda75..b5fde840e7 100644 --- a/test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java +++ b/test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java @@ -16,27 +16,33 @@ * limitations under the License. */ -package org.apache.cassandra.distributed; +package org.apache.cassandra.distributed.api; -import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.net.MessagingService; -import java.net.InetAddress; +import java.util.function.BiConsumer; -public class LegacyAdapter +public interface IMessageFilters { - private static final InetAddressAndPort broadcastAddressAndPort; - static + public interface Filter { - InetAddress address = FBUtilities.getBroadcastAddress(); - int port = DatabaseDescriptor.getStoragePort(); - broadcastAddressAndPort = InetAddressAndPort.getByAddressOverrideDefaults(address, port); + Filter restore(); + Filter drop(); } - public static InetAddressAndPort getBroadcastAddressAndPort() + public interface Builder { - return broadcastAddressAndPort; + Builder from(int ... nums); + Builder to(int ... nums); + Filter ready(); + Filter drop(); } + Builder verbs(MessagingService.Verb... verbs); + Builder allVerbs(); + void reset(); + + // internal + BiConsumer filter(BiConsumer applyIfNotFiltered); } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java new file mode 100644 index 0000000000..14ee1eeb74 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -0,0 +1,406 @@ +/* + * 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.impl; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import com.google.common.collect.Sets; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.distributed.api.ICoordinator; +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.distributed.api.IIsolatedExecutor; +import org.apache.cassandra.distributed.api.IListen; +import org.apache.cassandra.distributed.api.IMessage; +import org.apache.cassandra.distributed.api.IMessageFilters; +import org.apache.cassandra.distributed.api.ICluster; +import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.concurrent.SimpleCondition; + +/** + * AbstractCluster creates, initializes and manages Cassandra instances ({@link Instance}. + * + * All instances created under the same cluster will have a shared ClassLoader that'll preload + * common classes required for configuration and communication (byte buffers, primitives, config + * objects etc). Shared classes are listed in {@link InstanceClassLoader}. + * + * Each instance has its own class loader that will load logging, yaml libraries and all non-shared + * Cassandra package classes. The rule of thumb is that we'd like to have all Cassandra-specific things + * (unless explitily shared through the common classloader) on a per-classloader basis in order to + * allow creating more than one instance of DatabaseDescriptor and other Cassandra singletones. + * + * All actions (reading, writing, schema changes, etc) are executed by serializing lambda/runnables, + * transferring them to instance-specific classloaders, deserializing and running them there. Most of + * the things can be simply captured in closure or passed through `apply` method of the wrapped serializable + * function/callable. You can use {@link Instance#{applies|runs|consumes}OnInstance} for executing + * code on specific instance. + * + * Each instance has its own logger. Each instance log line will contain INSTANCE{instance_id}. + * + * As of today, messaging is faked by hooking into MessagingService, so we're not using usual Cassandra + * handlers for internode to have more control over it. Messaging is wired by passing verbs manually. + * coordinator-handling code and hooks to the callbacks can be found in {@link Coordinator}. + */ +public abstract class AbstractCluster implements ICluster, AutoCloseable +{ + // WARNING: we have this logger not (necessarily) for logging, but + // to ensure we have instantiated the main classloader's LoggerFactory (and any LogbackStatusListener) + // before we instantiate any for a new instance + private static final Logger logger = LoggerFactory.getLogger(AbstractCluster.class); + + private final File root; + private final ClassLoader sharedClassLoader; + + // mutated by starting/stopping a node + private final List instances; + private final Map instanceMap; + + // mutated by user-facing API + private final MessageFilters filters; + + protected class Wrapper extends DelegatingInvokableInstance implements IUpgradeableInstance + { + private final InstanceConfig config; + private volatile IInvokableInstance delegate; + private volatile Versions.Version version; + private volatile boolean isShutdown = true; + + protected IInvokableInstance delegate() + { + if (delegate == null) + delegate = newInstance(); + return delegate; + } + + public Wrapper(Versions.Version version, InstanceConfig config) + { + this.config = config; + this.version = version; + // we ensure there is always a non-null delegate, so that the executor may be used while the node is offline + this.delegate = newInstance(); + } + + private IInvokableInstance newInstance() + { + ClassLoader classLoader = new InstanceClassLoader(config.num(), version.classpath, sharedClassLoader); + return Instance.transferAdhoc((SerializableBiFunction)Instance::new, classLoader) + .apply(config, classLoader); + } + + public IInstanceConfig config() + { + return config; + } + + @Override + public synchronized void startup() + { + if (!isShutdown) + throw new IllegalStateException(); + delegate().startup(AbstractCluster.this); + isShutdown = false; + } + + @Override + public synchronized void shutdown() + { + if (isShutdown) + throw new IllegalStateException(); + isShutdown = true; + delegate.shutdown(); + delegate = null; + } + + @Override + public void receiveMessage(IMessage message) + { + IInvokableInstance delegate = this.delegate; + if (!isShutdown && delegate != null) // since we sync directly on the other node, we drop messages immediately if we are shutdown + delegate.receiveMessage(message); + } + + @Override + public synchronized void setVersion(Versions.Version version) + { + if (!isShutdown) + throw new IllegalStateException("Must be shutdown before version can be modified"); + // re-initialise + this.version = version; + if (delegate != null) + { + // we can have a non-null delegate even thought we are shutdown, if delegate() has been invoked since shutdown. + delegate.shutdown(); + delegate = null; + } + } + } + + protected AbstractCluster(File root, Versions.Version version, List configs, ClassLoader sharedClassLoader) + { + this.root = root; + this.sharedClassLoader = sharedClassLoader; + this.instances = new ArrayList<>(); + this.instanceMap = new HashMap<>(); + for (InstanceConfig config : configs) + { + I instance = newInstanceWrapper(version, config); + instances.add(instance); + // we use the config().broadcastAddressAndPort() here because we have not initialised the Instance + I prev = instanceMap.put(instance.broadcastAddressAndPort(), instance); + if (null != prev) + throw new IllegalStateException("Cluster cannot have multiple nodes with same InetAddressAndPort: " + instance.broadcastAddressAndPort() + " vs " + prev.broadcastAddressAndPort()); + } + this.filters = new MessageFilters(this); + } + + protected abstract I newInstanceWrapper(Versions.Version version, InstanceConfig config); + + /** + * WARNING: we index from 1 here, for consistency with inet address! + */ + public ICoordinator coordinator(int node) + { + return instances.get(node - 1).coordinator(); + } + /** + * WARNING: we index from 1 here, for consistency with inet address! + */ + public I get(int node) { return instances.get(node - 1); } + public I get(InetAddressAndPort addr) { return instanceMap.get(addr); } + + public int size() + { + return instances.size(); + } + public Stream stream() { return instances.stream(); } + public void forEach(IIsolatedExecutor.SerializableRunnable runnable) { forEach(i -> i.sync(runnable)); } + public void forEach(Consumer consumer) { instances.forEach(consumer); } + public void parallelForEach(IIsolatedExecutor.SerializableConsumer consumer, long timeout, TimeUnit units) + { + FBUtilities.waitOnFutures(instances.stream() + .map(i -> i.async(consumer).apply(i)) + .collect(Collectors.toList()), + timeout, units); + } + + + public IMessageFilters filters() { return filters; } + public MessageFilters.Builder verbs(MessagingService.Verb ... verbs) { return filters.verbs(verbs); } + + public void disableAutoCompaction(String keyspace) + { + forEach(() -> { + for (ColumnFamilyStore cs : Keyspace.open(keyspace).getColumnFamilyStores()) + cs.disableAutoCompaction(); + }); + } + + public void schemaChange(String query) + { + try (SchemaChangeMonitor monitor = new SchemaChangeMonitor()) + { + // execute the schema change + coordinator(1).execute(query, ConsistencyLevel.ALL); + monitor.waitForAgreement(); + } + } + + /** + * Will wait for a schema change AND agreement that occurs after it is created + * (and precedes the invocation to waitForAgreement) + * + * Works by simply checking if all UUIDs agree after any schema version change event, + * so long as the waitForAgreement method has been entered (indicating the change has + * taken place on the coordinator) + * + * This could perhaps be made a little more robust, but this should more than suffice. + */ + public class SchemaChangeMonitor implements AutoCloseable + { + final List cleanup; + volatile boolean schemaHasChanged; + final SimpleCondition agreement = new SimpleCondition(); + + public SchemaChangeMonitor() + { + this.cleanup = new ArrayList<>(instances.size()); + for (IInstance instance : instances) + cleanup.add(instance.listen().schema(this::signal)); + } + + private void signal() + { + if (schemaHasChanged && 1 == instances.stream().map(IInstance::schemaVersion).distinct().count()) + agreement.signalAll(); + } + + @Override + public void close() + { + for (IListen.Cancel cancel : cleanup) + cancel.cancel(); + } + + public void waitForAgreement() + { + schemaHasChanged = true; + signal(); + try + { + if (!agreement.await(1L, TimeUnit.MINUTES)) + throw new InterruptedException(); + } + catch (InterruptedException e) + { + throw new IllegalStateException("Schema agreement not reached"); + } + } + } + + public void schemaChange(String statement, int instance) + { + get(instance).schemaChangeInternal(statement); + } + + void startup() + { + parallelForEach(I::startup, 0, null); + } + + protected interface Factory> + { + C newCluster(File root, Versions.Version version, List configs, ClassLoader sharedClassLoader); + } + + protected static > C + create(int nodeCount, Factory factory) throws Throwable + { + return create(nodeCount, Files.createTempDirectory("dtests").toFile(), factory); + } + + protected static > C + create(int nodeCount, File root, Factory factory) + { + return create(nodeCount, Versions.CURRENT, root, factory); + } + + protected static > C + create(int nodeCount, Versions.Version version, Factory factory) throws IOException + { + return create(nodeCount, version, Files.createTempDirectory("dtests").toFile(), factory); + } + + protected static > C + create(int nodeCount, Versions.Version version, File root, Factory factory) + { + root.mkdirs(); + setupLogging(root); + + ClassLoader sharedClassLoader = Thread.currentThread().getContextClassLoader(); + + List configs = new ArrayList<>(); + long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount); + for (int i = 0 ; i < nodeCount ; ++i) + { + InstanceConfig config = InstanceConfig.generate(i + 1, root, String.valueOf(token)); + configs.add(config); + token += increment; + } + + C cluster = factory.newCluster(root, version, configs, sharedClassLoader); + cluster.startup(); + return cluster; + } + + private static void setupLogging(File root) + { + try + { + String testConfPath = "test/conf/logback-dtest.xml"; + Path logConfPath = Paths.get(root.getPath(), "/logback-dtest.xml"); + if (!logConfPath.toFile().exists()) + { + Files.copy(new File(testConfPath).toPath(), + logConfPath); + } + System.setProperty("logback.configurationFile", "file://" + logConfPath); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + @Override + public void close() + { + parallelForEach(IInstance::shutdown, 1L, TimeUnit.MINUTES); + + // 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 + private void withThreadLeakCheck(List> futures) + { + FBUtilities.waitOnFutures(futures); + + Set threadSet = Thread.getAllStackTraces().keySet(); + threadSet = Sets.difference(threadSet, Collections.singletonMap(Thread.currentThread(), null).keySet()); + if (!threadSet.isEmpty()) + { + for (Thread thread : threadSet) + { + System.out.println(thread); + System.out.println(Arrays.toString(thread.getStackTrace())); + } + throw new RuntimeException(String.format("Not all threads have shut down. %d threads are still running: %s", threadSet.size(), threadSet)); + } + } + +} + diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java new file mode 100644 index 0000000000..655ceb82d0 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java @@ -0,0 +1,75 @@ +/* + * 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.impl; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; + +import org.apache.cassandra.cql3.CQLStatement; +import org.apache.cassandra.cql3.QueryOptions; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.distributed.api.ICoordinator; +import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.service.QueryState; +import org.apache.cassandra.transport.Server; +import org.apache.cassandra.transport.messages.ResultMessage; +import org.apache.cassandra.utils.ByteBufferUtil; + +public class Coordinator implements ICoordinator +{ + final Instance instance; + public Coordinator(Instance instance) + { + this.instance = instance; + } + + @Override + public Object[][] execute(String query, Enum consistencyLevelOrigin, Object... boundValues) + { + return instance.sync(() -> { + ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name()); + CQLStatement prepared = QueryProcessor.getStatement(query, ClientState.forInternalCalls()).statement; + List boundBBValues = new ArrayList<>(); + for (Object boundValue : boundValues) + { + boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue)); + } + + ResultMessage res = prepared.execute(QueryState.forInternalCalls(), + QueryOptions.create(consistencyLevel, + boundBBValues, + false, + 10, + null, + null, + Server.CURRENT_VERSION)); + + if (res != null && res.kind == ResultMessage.Kind.ROWS) + { + return RowUtil.toObjects((ResultMessage.Rows) res); + } + else + { + return new Object[][]{}; + } + }).call(); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java b/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java new file mode 100644 index 0000000000..68f957baff --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java @@ -0,0 +1,196 @@ +/* + * 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.impl; + +import java.io.Serializable; +import java.util.UUID; +import java.util.concurrent.Future; +import java.util.function.BiConsumer; +import java.util.function.BiFunction; +import java.util.function.Consumer; +import java.util.function.Function; + +import org.apache.cassandra.distributed.api.ICluster; +import org.apache.cassandra.distributed.api.ICoordinator; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.distributed.api.IListen; +import org.apache.cassandra.distributed.api.IMessage; +import org.apache.cassandra.locator.InetAddressAndPort; + +public abstract class DelegatingInvokableInstance implements IInvokableInstance +{ + protected abstract IInvokableInstance delegate(); + + @Override + public E transfer(E object) + { + return delegate().transfer(object); + } + + @Override + public InetAddressAndPort broadcastAddressAndPort() + { + return delegate().broadcastAddressAndPort(); + } + + @Override + public Object[][] executeInternal(String query, Object... args) + { + return delegate().executeInternal(query, args); + } + + @Override + public UUID schemaVersion() + { + return delegate().schemaVersion(); + } + + @Override + public void startup() + { + throw new UnsupportedOperationException(); + } + + @Override + public void schemaChangeInternal(String query) + { + delegate().schemaChangeInternal(query); + } + + @Override + public IInstanceConfig config() + { + return delegate().config(); + } + + @Override + public ICoordinator coordinator() + { + // TODO: stash and clear coordinator on startup/shutdown? + return delegate().coordinator(); + } + + public IListen listen() + { + return delegate().listen(); + } + + @Override + public void shutdown() + { + delegate().shutdown(); + } + + @Override + public void startup(ICluster cluster) + { + delegate().startup(cluster); + } + + @Override + public void receiveMessage(IMessage message) + { + delegate().receiveMessage(message); + } + + @Override + public CallableNoExcept> async(CallableNoExcept call) + { + return delegate().async(call); + } + + @Override + public CallableNoExcept sync(CallableNoExcept call) + { + return delegate().sync(call); + } + + @Override + public CallableNoExcept> async(Runnable run) + { + return delegate().async(run); + } + + @Override + public Runnable sync(Runnable run) + { + return delegate().sync(run); + } + + @Override + public Function> async(Consumer consumer) + { + return delegate().async(consumer); + } + + @Override + public Consumer sync(Consumer consumer) + { + return delegate().sync(consumer); + } + + @Override + public BiFunction> async(BiConsumer consumer) + { + return delegate().async(consumer); + } + + @Override + public BiConsumer sync(BiConsumer consumer) + { + return delegate().sync(consumer); + } + + @Override + public Function> async(Function f) + { + return delegate().async(f); + } + + @Override + public Function sync(Function f) + { + return delegate().sync(f); + } + + @Override + public BiFunction> async(BiFunction f) + { + return delegate().async(f); + } + + @Override + public BiFunction sync(BiFunction f) + { + return delegate().sync(f); + } + + @Override + public TriFunction> async(TriFunction f) + { + return delegate().async(f); + } + + @Override + public TriFunction sync(TriFunction f) + { + return delegate().sync(f); + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/impl/IInvokableInstance.java b/test/distributed/org/apache/cassandra/distributed/impl/IInvokableInstance.java new file mode 100644 index 0000000000..6fe5891e78 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/IInvokableInstance.java @@ -0,0 +1,68 @@ +/* + * 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.impl; + +import java.io.Serializable; +import java.util.concurrent.Future; +import java.util.function.BiConsumer; +import java.util.function.BiFunction; +import java.util.function.Consumer; +import java.util.function.Function; + +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.distributed.api.IIsolatedExecutor; + +/** + * This version is only supported for a Cluster running the same code as the test environment, and permits + * ergonomic cross-node behaviours, without editing the cross-version API. + * + * A lambda can be written tto be invoked on any or all of the nodes. + * + * The reason this cannot (easily) be made cross-version is that the lambda is tied to the declaring class, which will + * not be the same in the alternate version. Even were it not, there would likely be a runtime linkage error given + * any code divergence. + */ +public interface IInvokableInstance extends IInstance +{ + public default CallableNoExcept> asyncCallsOnInstance(SerializableCallable call) { return async(transfer(call)); } + public default CallableNoExcept callsOnInstance(SerializableCallable call) { return sync(transfer(call)); } + public default O callOnInstance(SerializableCallable call) { return callsOnInstance(call).call(); } + + public default CallableNoExcept> asyncRunsOnInstance(SerializableRunnable run) { return async(transfer(run)); } + public default Runnable runsOnInstance(SerializableRunnable run) { return sync(transfer(run)); } + public default void runOnInstance(SerializableRunnable run) { runsOnInstance(run).run(); } + + public default Function> asyncAcceptsOnInstance(SerializableConsumer consumer) { return async(transfer(consumer)); } + public default Consumer acceptsOnInstance(SerializableConsumer consumer) { return sync(transfer(consumer)); } + + public default BiFunction> asyncAcceptsOnInstance(SerializableBiConsumer consumer) { return async(transfer(consumer)); } + public default BiConsumer acceptsOnInstance(SerializableBiConsumer consumer) { return sync(transfer(consumer)); } + + public default Function> asyncAppliesOnInstance(SerializableFunction f) { return async(transfer(f)); } + public default Function appliesOnInstance(SerializableFunction f) { return sync(transfer(f)); } + + public default BiFunction> asyncAppliesOnInstance(SerializableBiFunction f) { return async(transfer(f)); } + public default BiFunction appliesOnInstance(SerializableBiFunction f) { return sync(transfer(f)); } + + public default TriFunction> asyncAppliesOnInstance(SerializableTriFunction f) { return async(transfer(f)); } + public default TriFunction appliesOnInstance(SerializableTriFunction f) { return sync(transfer(f)); } + + public E transfer(E object); + +} diff --git a/test/distributed/org/apache/cassandra/distributed/impl/IUpgradeableInstance.java b/test/distributed/org/apache/cassandra/distributed/impl/IUpgradeableInstance.java new file mode 100644 index 0000000000..3eb3657d4d --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/IUpgradeableInstance.java @@ -0,0 +1,28 @@ +/* + * 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.impl; + +import org.apache.cassandra.distributed.api.IInstance; + +// this lives outside the api package so that we do not have to worry about inter-version compatibility +public interface IUpgradeableInstance extends IInstance +{ + // only to be invoked while the node is shutdown! + public void setVersion(Versions.Version version); +} diff --git a/test/distributed/org/apache/cassandra/distributed/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java similarity index 55% rename from test/distributed/org/apache/cassandra/distributed/Instance.java rename to test/distributed/org/apache/cassandra/distributed/impl/Instance.java index ccf79614db..656f3f6f3c 100644 --- a/test/distributed/org/apache/cassandra/distributed/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -16,12 +16,11 @@ * limitations under the License. */ -package org.apache.cassandra.distributed; +package org.apache.cassandra.distributed.impl; import java.io.File; import java.io.IOException; import java.net.InetAddress; -import java.net.UnknownHostException; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -42,7 +41,6 @@ import org.apache.cassandra.concurrent.SharedExecutorPool; import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.config.ParameterizedClass; import org.apache.cassandra.config.Schema; import org.apache.cassandra.cql3.CQLStatement; import org.apache.cassandra.cql3.QueryOptions; @@ -56,6 +54,11 @@ import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.distributed.api.ICoordinator; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.distributed.api.IListen; +import org.apache.cassandra.distributed.api.IMessage; +import org.apache.cassandra.distributed.api.ICluster; import org.apache.cassandra.gms.ApplicationState; import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.gms.VersionedValue; @@ -65,13 +68,12 @@ 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.locator.SimpleSeedProvider; -import org.apache.cassandra.locator.SimpleSnitch; import org.apache.cassandra.net.IMessageSink; import org.apache.cassandra.net.MessageDeliveryTask; import org.apache.cassandra.net.MessageIn; import org.apache.cassandra.net.MessageOut; import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.schema.LegacySchemaMigrator; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.PendingRangeCalculatorService; import org.apache.cassandra.service.QueryState; @@ -82,44 +84,69 @@ import org.apache.cassandra.utils.Throwables; import org.apache.cassandra.utils.concurrent.Ref; import org.apache.cassandra.utils.memory.BufferPool; -public class Instance extends InvokableInstance +public class Instance extends IsolatedExecutor implements IInvokableInstance { - public final InstanceConfig config; + public final IInstanceConfig config; - public Instance(InstanceConfig config, ClassLoader classLoader) + // should never be invoked directly, so that it is instantiated on other class loader; + // only visible for inheritance + Instance(IInstanceConfig config, ClassLoader classLoader) { - super("node" + config.num, classLoader); + super("node" + config.num(), classLoader); this.config = config; + InstanceIDDefiner.setInstanceId(config.num()); + FBUtilities.setBroadcastInetAddress(config.broadcastAddressAndPort().address); } - public InetAddressAndPort getBroadcastAddress() { return callOnInstance(LegacyAdapter::getBroadcastAddressAndPort); } + public IInstanceConfig config() + { + return config; + } + + public ICoordinator coordinator() + { + return new Coordinator(this); + } + + public IListen listen() + { + return new Listen(this); + } + + @Override + public InetAddressAndPort broadcastAddressAndPort() { return config.broadcastAddressAndPort(); } public Object[][] executeInternal(String query, Object... args) { - return callOnInstance(() -> - { + return sync(() -> { ParsedStatement.Prepared prepared = QueryProcessor.prepareInternal(query); ResultMessage result = prepared.statement.executeInternal(QueryProcessor.internalQueryState(), - QueryProcessor.makeInternalOptions(prepared, args)); + QueryProcessor.makeInternalOptions(prepared, args)); if (result instanceof ResultMessage.Rows) return RowUtil.toObjects((ResultMessage.Rows)result); else return null; - }); + }).call(); } - public UUID getSchemaVersion() + @Override + public UUID schemaVersion() { - // we do not use method reference syntax here, because we need to invoke on the node-local schema instance + // we do not use method reference syntax here, because we need to sync on the node-local schema instance //noinspection Convert2MethodRef - return callOnInstance(() -> Schema.instance.getVersion()); + return Schema.instance.getVersion(); } - public void schemaChange(String query) + public void startup() { - runOnInstance(() -> - { + throw new UnsupportedOperationException(); + } + + @Override + public void schemaChangeInternal(String query) + { + sync(() -> { try { ClientState state = ClientState.forInternalCalls(); @@ -136,30 +163,33 @@ public class Instance extends InvokableInstance { throw new RuntimeException("Error setting schema for test (query was: " + query + ")", e); } - }); + }).run(); } - private void registerMockMessaging(TestCluster cluster) + private void registerMockMessaging(ICluster cluster) { - BiConsumer deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message); - BiConsumer deliverToInstanceIfNotFiltered = cluster.filters().filter(deliverToInstance); + BiConsumer deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message); + BiConsumer deliverToInstanceIfNotFiltered = cluster.filters().filter(deliverToInstance); Map addressAndPortMap = new HashMap<>(); - cluster.stream().map(Instance::getBroadcastAddress).forEach(addressAndPort -> { - if (null != addressAndPortMap.put(addressAndPort.address, addressAndPort)) - throw new IllegalStateException("This version of Cassandra does not support multiple nodes with the same InetAddress"); + cluster.stream().forEach(instance -> { + InetAddressAndPort addressAndPort = instance.broadcastAddressAndPort(); + if (!addressAndPort.equals(instance.config().broadcastAddressAndPort())) + throw new IllegalStateException("addressAndPort mismatch: " + addressAndPort + " vs " + instance.config().broadcastAddressAndPort()); + InetAddressAndPort prev = addressAndPortMap.put(addressAndPort.address, addressAndPort); + if (null != prev) + throw new IllegalStateException("This version of Cassandra does not support multiple nodes with the same InetAddress: " + addressAndPort + " vs " + prev); }); - acceptsOnInstance((BiConsumer deliver) -> - MessagingService.instance().addMessageSink(new MessageDeliverySink(deliver, addressAndPortMap::get)) - ).accept(deliverToInstanceIfNotFiltered); + MessagingService.instance().addMessageSink( + new MessageDeliverySink(deliverToInstanceIfNotFiltered, addressAndPortMap::get)); } - private static class MessageDeliverySink implements IMessageSink + private class MessageDeliverySink implements IMessageSink { - private final BiConsumer deliver; + private final BiConsumer deliver; private final Function lookupAddressAndPort; - MessageDeliverySink(BiConsumer deliver, Function lookupAddressAndPort) + MessageDeliverySink(BiConsumer deliver, Function lookupAddressAndPort) { this.deliver = deliver; this.lookupAddressAndPort = lookupAddressAndPort; @@ -169,7 +199,8 @@ public class Instance extends InvokableInstance { try (DataOutputBuffer out = new DataOutputBuffer(1024)) { - InetAddressAndPort from = LegacyAdapter.getBroadcastAddressAndPort(); + InetAddressAndPort from = broadcastAddressAndPort(); + assert from.equals(lookupAddressAndPort.apply(messageOut.from)); InetAddressAndPort toFull = lookupAddressAndPort.apply(to); messageOut.serialize(out, MessagingService.current_version); deliver.accept(toFull, new Message(messageOut.verb.ordinal(), out.toByteArray(), id, MessagingService.current_version, from)); @@ -188,158 +219,146 @@ public class Instance extends InvokableInstance } } - private void receiveMessage(Message message) + public void receiveMessage(IMessage message) { - acceptsOnInstance((Message m) -> - { - try (DataInputBuffer in = new DataInputBuffer(m.bytes)) + sync(() -> { + try (DataInputBuffer in = new DataInputBuffer(message.bytes())) { - MessageIn messageIn = MessageIn.read(in, m.version, m.id); - Runnable deliver = new MessageDeliveryTask(messageIn, m.id, System.currentTimeMillis(), false); + MessageIn messageIn = MessageIn.read(in, message.version(), message.id()); + Runnable deliver = new MessageDeliveryTask(messageIn, message.id(), System.currentTimeMillis(), false); deliver.run(); } catch (Throwable t) { - throw new RuntimeException("Exception occurred on the node " + LegacyAdapter.getBroadcastAddressAndPort(), t); + throw new RuntimeException("Exception occurred on node " + broadcastAddressAndPort(), t); } - - }).accept(message); + }).run(); } - void launch(TestCluster cluster) + @Override + public void startup(ICluster cluster) { - try - { - mkdirs(); - int id = config.num; - runOnInstance(() -> InstanceIDDefiner.instanceId = id); // for logging + sync(() -> { + try + { + mkdirs(); - startup(); - initializeRing(cluster); - registerMockMessaging(cluster); - } - catch (Throwable t) - { - if (t instanceof RuntimeException) - throw (RuntimeException) t; - throw new RuntimeException(t); - } + Config.setOverrideLoadConfig(() -> loadConfig(config)); + DatabaseDescriptor.setDaemonInitialized(); + DatabaseDescriptor.createAllDirectories(); + + // We need to persist this as soon as possible after startup checks. + // This should be the first write to SystemKeyspace (CASSANDRA-11742) + SystemKeyspace.persistLocalMetadata(); + LegacySchemaMigrator.migrate(); + + try + { + // load schema from disk + Schema.instance.loadFromDisk(); + } + catch (Exception e) + { + throw e; + } + + Keyspace.setInitialized(); + + // Replay any CommitLogSegments found on disk + try + { + CommitLog.instance.recover(); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + + initializeRing(cluster); + registerMockMessaging(cluster); + + SystemKeyspace.finishStartup(); + + if (!FBUtilities.getBroadcastAddress().equals(broadcastAddressAndPort().address)) + throw new IllegalStateException(); + if (DatabaseDescriptor.getStoragePort() != broadcastAddressAndPort().port) + throw new IllegalStateException(); + } + catch (Throwable t) + { + if (t instanceof RuntimeException) + throw (RuntimeException) t; + throw new RuntimeException(t); + } + }).run(); } private void mkdirs() { - new File(config.saved_caches_directory).mkdirs(); - new File(config.hints_directory).mkdirs(); - new File(config.commitlog_directory).mkdirs(); - for (String dir : config.data_file_directories) + new File(config.getString("saved_caches_directory")).mkdirs(); + new File(config.getString("hints_directory")).mkdirs(); + new File(config.getString("commitlog_directory")).mkdirs(); + for (String dir : (String[]) config.get("data_file_directories")) new File(dir).mkdirs(); } - private void startup() - { - acceptsOnInstance((InstanceConfig config) -> - { - Config.setOverrideLoadConfig(() -> loadConfig(config)); - DatabaseDescriptor.setDaemonInitialized(); - DatabaseDescriptor.createAllDirectories(); - Keyspace.setInitialized(); - SystemKeyspace.persistLocalMetadata(); - }).accept(config); - } - - - public static Config loadConfig(InstanceConfig overrides) + private static Config loadConfig(IInstanceConfig overrides) { Config config = new Config(); - // Defaults - config.commitlog_sync = Config.CommitLogSync.batch; - config.endpoint_snitch = SimpleSnitch.class.getName(); - config.seed_provider = new ParameterizedClass(SimpleSeedProvider.class.getName(), - Collections.singletonMap("seeds", "127.0.0.1:7010")); - // Overrides - config.partitioner = overrides.partitioner; - config.broadcast_address = overrides.broadcast_address; - config.listen_address = overrides.listen_address; - config.broadcast_rpc_address = overrides.broadcast_rpc_address; - config.rpc_address = overrides.rpc_address; - config.saved_caches_directory = overrides.saved_caches_directory; - config.data_file_directories = overrides.data_file_directories; - config.commitlog_directory = overrides.commitlog_directory; - config.hints_directory = overrides.hints_directory; - config.concurrent_writes = overrides.concurrent_writes; - config.concurrent_counter_writes = overrides.concurrent_counter_writes; - config.concurrent_materialized_view_writes = overrides.concurrent_materialized_view_writes; - config.concurrent_reads = overrides.concurrent_reads; - config.memtable_flush_writers = overrides.memtable_flush_writers; - config.concurrent_compactors = overrides.concurrent_compactors; - config.memtable_heap_space_in_mb = overrides.memtable_heap_space_in_mb; - config.initial_token = overrides.initial_token; - - // legacy config options we need to specify - config.commitlog_sync_batch_window_in_ms = 1.0; + overrides.propagate(config); return config; } - private void initializeRing(TestCluster cluster) + private void initializeRing(ICluster cluster) { // This should be done outside instance in order to avoid serializing config - String partitionerName = config.partitioner; + String partitionerName = config.getString("partitioner"); List initialTokens = new ArrayList<>(); List hosts = new ArrayList<>(); List hostIds = new ArrayList<>(); for (int i = 1 ; i <= cluster.size() ; ++i) { - InstanceConfig config = cluster.get(i).config; - initialTokens.add(config.initial_token); - try - { - hosts.add(InetAddressAndPort.getByName(config.broadcast_address)); - } - catch (UnknownHostException e) - { - throw new RuntimeException(e); - } - hostIds.add(config.hostId); + IInstanceConfig config = cluster.get(i).config(); + initialTokens.add(config.getString("initial_token")); + hosts.add(config.broadcastAddressAndPort()); + hostIds.add(config.hostId()); } - runOnInstance(() -> + try { - try - { - IPartitioner partitioner = FBUtilities.newPartitioner(partitionerName); - StorageService storageService = StorageService.instance; - List tokens = new ArrayList<>(); - for (String token : initialTokens) - tokens.add(partitioner.getTokenFactory().fromString(token)); + IPartitioner partitioner = FBUtilities.newPartitioner(partitionerName); + StorageService storageService = StorageService.instance; + List tokens = new ArrayList<>(); + for (String token : initialTokens) + tokens.add(partitioner.getTokenFactory().fromString(token)); - 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)); - MessagingService.instance().setVersion(ep.address, MessagingService.current_version); - } - - // check that all nodes are in token metadata - for (int i = 0; i < tokens.size(); ++i) - assert storageService.getTokenMetadata().isMember(hosts.get(i).address); - } - catch (Throwable e) // UnknownHostException + for (int i = 0; i < tokens.size(); i++) { - throw new RuntimeException(e); + 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)); + MessagingService.instance().setVersion(ep.address, MessagingService.current_version); } - }); + + // check that all nodes are in token metadata + for (int i = 0; i < tokens.size(); ++i) + assert storageService.getTokenMetadata().isMember(hosts.get(i).address); + } + catch (Throwable e) // UnknownHostException + { + throw new RuntimeException(e); + } } - void shutdown() + public void shutdown() { - acceptsOnInstance((ExecutorService executor) -> { + sync((ExecutorService executor) -> { Throwable error = null; error = parallelRun(error, executor, Gossiper.instance::stop, @@ -368,9 +387,9 @@ public class Instance extends InvokableInstance ); LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); loggerContext.stop(); + super.shutdown(); Throwables.maybeFail(error); }).accept(isolatedExecutor); - super.shutdown(); } private static Throwable parallelRun(Throwable accumulate, ExecutorService runOn, ThrowingRunnable ... runnables) diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java new file mode 100644 index 0000000000..2b29626ae1 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java @@ -0,0 +1,103 @@ +/* + * 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.impl; + +import com.google.common.base.Predicate; +import org.apache.cassandra.config.ParameterizedClass; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.utils.Pair; + +import java.net.URL; +import java.net.URLClassLoader; +import java.util.Arrays; +import java.util.Set; +import java.util.stream.Collectors; + +public class InstanceClassLoader extends URLClassLoader +{ + // Classes that have to be shared between instances, for configuration or returning values + private static final Set sharedClassNames = Arrays.stream(new Class[] + { + Pair.class, + InetAddressAndPort.class, + ParameterizedClass.class, + IInvokableInstance.class + }) + .map(Class::getName) + .collect(Collectors.toSet()); + + private static final Predicate sharePackage = name -> + name.startsWith("org.apache.cassandra.distributed.api.") + || name.startsWith("sun.") + || name.startsWith("oracle.") + || name.startsWith("com.sun.") + || name.startsWith("com.oracle.") + || name.startsWith("java.") + || name.startsWith("javax.") + || name.startsWith("jdk.") + || name.startsWith("netscape.") + || name.startsWith("org.xml.sax."); + + private static final Predicate shareClass = name -> sharePackage.apply(name) || sharedClassNames.contains(name); + + public static interface Factory + { + InstanceClassLoader create(int id, URL[] urls, ClassLoader sharedClassLoader); + } + + private final ClassLoader sharedClassLoader; + + InstanceClassLoader(int id, URL[] urls, ClassLoader sharedClassLoader) + { + super(urls, null); + this.sharedClassLoader = sharedClassLoader; + } + + @Override + public Class loadClass(String name) throws ClassNotFoundException + { + if (shareClass.apply(name)) + return sharedClassLoader.loadClass(name); + + return loadClassInternal(name); + } + + Class loadClassInternal(String name) throws ClassNotFoundException + { + synchronized (getClassLoadingLock(name)) + { + // First, check if the class has already been loaded + Class c = findLoadedClass(name); + + if (c == null) + c = findClass(name); + + return c; + } + } + + /** + * @return true iff this class was loaded by an InstanceClassLoader, and as such is used by a dtest node + */ + public static boolean wasLoadedByAnInstanceClassLoader(Class clazz) + { + return clazz.getClassLoader().getClass().getName().equals(InstanceClassLoader.class.getName()); + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java new file mode 100644 index 0000000000..786e2d7e8e --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java @@ -0,0 +1,233 @@ +/* + * 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.impl; + +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.ParameterizedClass; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.locator.SimpleSeedProvider; +import org.apache.cassandra.locator.SimpleSnitch; + +import java.io.File; +import java.lang.reflect.Field; +import java.net.UnknownHostException; +import java.util.Arrays; +import java.util.Collections; +import java.util.Map; +import java.util.TreeMap; +import java.util.UUID; + +public class InstanceConfig implements IInstanceConfig +{ + private static final Object NULL = new Object(); + + public final int num; + public int num() { return num; } + + public final UUID hostId; + public UUID hostId() { return hostId; } + private final Map params = new TreeMap<>(); + + private volatile InetAddressAndPort broadcastAddressAndPort; + + @Override + public InetAddressAndPort broadcastAddressAndPort() + { + if (broadcastAddressAndPort == null) + { + try + { + broadcastAddressAndPort = InetAddressAndPort.getByNameOverrideDefaults(getString("broadcast_address"), getInt("storage_port")); + } + catch (UnknownHostException e) + { + throw new IllegalStateException(e); + } + } + return broadcastAddressAndPort; + } + + private InstanceConfig(int num, + String broadcast_address, + String listen_address, + String broadcast_rpc_address, + String rpc_address, + String saved_caches_directory, + String[] data_file_directories, + String commitlog_directory, + String hints_directory, +// String cdc_directory, + String initial_token) + { + this.num = num; + this.hostId = java.util.UUID.randomUUID(); + this .set("broadcast_address", broadcast_address) + .set("listen_address", listen_address) + .set("broadcast_rpc_address", broadcast_rpc_address) + .set("rpc_address", rpc_address) + .set("saved_caches_directory", saved_caches_directory) + .set("data_file_directories", data_file_directories) + .set("commitlog_directory", commitlog_directory) + .set("hints_directory", hints_directory) +// .set("cdc_directory", cdc_directory) + .set("initial_token", initial_token) + .set("partitioner", "org.apache.cassandra.dht.Murmur3Partitioner") + .set("concurrent_writes", 2) + .set("concurrent_counter_writes", 2) + .set("concurrent_materialized_view_writes", 2) + .set("concurrent_reads", 2) + .set("memtable_flush_writers", 1) + .set("concurrent_compactors", 1) + .set("memtable_heap_space_in_mb", 10) + .set("commitlog_sync", "batch") + .set("storage_port", 7010) + .set("endpoint_snitch", SimpleSnitch.class.getName()) + .set("seed_provider", new ParameterizedClass(SimpleSeedProvider.class.getName(), + Collections.singletonMap("seeds", "127.0.0.1"))) + // legacy parameters + .forceSet("commitlog_sync_batch_window_in_ms", 1.0); + } + + private InstanceConfig(InstanceConfig copy) + { + this.num = copy.num; + this.params.putAll(copy.params); + this.hostId = copy.hostId; + } + + public InstanceConfig set(String fieldName, Object value) + { + if (value == null) + value = NULL; + + // test value + propagate(new Config(), fieldName, value, false); + params.put(fieldName, value); + return this; + } + + private InstanceConfig forceSet(String fieldName, Object value) + { + if (value == null) + value = NULL; + + // test value + params.put(fieldName, value); + return this; + } + + public void propagateIfSet(Object writeToConfig, String fieldName) + { + if (params.containsKey(fieldName)) + propagate(writeToConfig, fieldName, params.get(fieldName), true); + } + + public void propagate(Object writeToConfig) + { + for (Map.Entry e : params.entrySet()) + propagate(writeToConfig, e.getKey(), e.getValue(), true); + } + + private void propagate(Object writeToConfig, String fieldName, Object value, boolean ignoreMissing) + { + if (value == NULL) + value = null; + + Class configClass = writeToConfig.getClass(); + Field valueField; + try + { + valueField = configClass.getDeclaredField(fieldName); + } + catch (NoSuchFieldException e) + { + if (!ignoreMissing) + throw new IllegalStateException(e); + return; + } + + if (valueField.getType().isEnum() && value instanceof String) + { + String test = (String) value; + value = Arrays.stream(valueField.getType().getEnumConstants()) + .filter(e -> ((Enum)e).name().equals(test)) + .findFirst() + .get(); + } + try + { + valueField.set(writeToConfig, value); + } + catch (IllegalAccessException e) + { + throw new IllegalStateException(e); + } + catch (IllegalArgumentException e) + { + throw new IllegalStateException(e); + } + } + + public Object get(String name) + { + return params.get(name); + } + + public int getInt(String name) + { + return (Integer)params.get(name); + } + + public String getString(String name) + { + return (String)params.get(name); + } + + public static InstanceConfig generate(int nodeNum, File root, String token) + { + return new InstanceConfig(nodeNum, + "127.0.0." + nodeNum, + "127.0.0." + nodeNum, + "127.0.0." + nodeNum, + "127.0.0." + nodeNum, + String.format("%s/node%d/saved_caches", root, nodeNum), + new String[] { String.format("%s/node%d/data", root, nodeNum) }, + String.format("%s/node%d/commitlog", root, nodeNum), + String.format("%s/node%d/hints", root, nodeNum), +// String.format("%s/node%d/cdc", root, nodeNum), + token); + } + + public InstanceConfig forVersion(Versions.Major major) + { + switch (major) + { + case v4: return this; + default: return new InstanceConfig(this) + .set("seed_provider", new ParameterizedClass(SimpleSeedProvider.class.getName(), + Collections.singletonMap("seeds", "127.0.0.1"))); + } + } + + public String toString() + { + return params.toString(); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceIDDefiner.java similarity index 76% rename from test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java rename to test/distributed/org/apache/cassandra/distributed/impl/InstanceIDDefiner.java index 11677487fa..d32bd7759e 100644 --- a/test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceIDDefiner.java @@ -16,9 +16,10 @@ * limitations under the License. */ -package org.apache.cassandra.distributed; +package org.apache.cassandra.distributed.impl; import ch.qos.logback.core.PropertyDefinerBase; +import org.apache.cassandra.concurrent.NamedThreadFactory; /** * Used by logback to find/define property value, see logback-dtest.xml @@ -26,13 +27,15 @@ import ch.qos.logback.core.PropertyDefinerBase; public class InstanceIDDefiner extends PropertyDefinerBase { // Instantiated per classloader, set by Instance - public static int instanceId = -1; + private static volatile String instanceId = "
"; + public static void setInstanceId(int id) + { + instanceId = "node" + id; + NamedThreadFactory.setGlobalPrefix("node" + id + "_"); + } public String getPropertyValue() { - if (instanceId == -1) - return "
"; - else - return "INSTANCE" + instanceId; + return instanceId; } } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java new file mode 100644 index 0000000000..863164f051 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java @@ -0,0 +1,165 @@ +/* + * 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.impl; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.function.BiConsumer; +import java.util.function.BiFunction; +import java.util.function.Consumer; +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 +{ + final ExecutorService isolatedExecutor; + private final ClassLoader classLoader; + private final Method deserializeOnInstance; + + IsolatedExecutor(String name, ClassLoader classLoader) + { + this.isolatedExecutor = Executors.newCachedThreadPool(new NamedThreadFactory("isolatedExecutor", Thread.NORM_PRIORITY, classLoader, new ThreadGroup(name))); + this.classLoader = classLoader; + this.deserializeOnInstance = lookupDeserializeOneObject(classLoader); + } + + public void shutdown() + { + isolatedExecutor.shutdown(); + } + + public CallableNoExcept> async(CallableNoExcept call) { return () -> isolatedExecutor.submit(call); } + public CallableNoExcept sync(CallableNoExcept call) { return () -> waitOn(async(call).call()); } + + public CallableNoExcept> async(Runnable run) { return () -> isolatedExecutor.submit(run); } + public Runnable sync(Runnable run) { return () -> waitOn(async(run).call()); } + + public Function> async(Consumer consumer) { return (a) -> isolatedExecutor.submit(() -> consumer.accept(a)); } + public Consumer sync(Consumer consumer) { return (a) -> waitOn(async(consumer).apply(a)); } + + public BiFunction> async(BiConsumer consumer) { return (a, b) -> isolatedExecutor.submit(() -> consumer.accept(a, b)); } + public BiConsumer sync(BiConsumer consumer) { return (a, b) -> waitOn(async(consumer).apply(a, b)); } + + public Function> async(Function f) { return (a) -> isolatedExecutor.submit(() -> f.apply(a)); } + public Function sync(Function f) { return (a) -> waitOn(async(f).apply(a)); } + + public BiFunction> async(BiFunction f) { return (a, b) -> isolatedExecutor.submit(() -> f.apply(a, b)); } + public BiFunction sync(BiFunction f) { return (a, b) -> waitOn(async(f).apply(a, b)); } + + public TriFunction> async(TriFunction f) { return (a, b, c) -> isolatedExecutor.submit(() -> f.apply(a, b, c)); } + public TriFunction sync(TriFunction f) { return (a, b, c) -> waitOn(async(f).apply(a, b, c)); } + + public E transfer(E object) + { + return (E) transferOneObject(object, classLoader, deserializeOnInstance); + } + + static E transferAdhoc(E object, ClassLoader classLoader) + { + return transferOneObject(object, classLoader, lookupDeserializeOneObject(classLoader)); + } + + private static E transferOneObject(E object, ClassLoader classLoader, Method deserializeOnInstance) + { + byte[] bytes = serializeOneObject(object); + try + { + Object onInstance = deserializeOnInstance.invoke(null, bytes); + if (onInstance.getClass().getClassLoader() != classLoader) + throw new IllegalStateException(onInstance + " seemingly from wrong class loader: " + onInstance.getClass().getClassLoader() + ", but expected " + classLoader); + + return (E) onInstance; + } + catch (IllegalAccessException | InvocationTargetException e) + { + throw new RuntimeException(e); + } + } + + private static Method lookupDeserializeOneObject(ClassLoader classLoader) + { + try + { + return classLoader.loadClass(IsolatedExecutor.class.getName()).getDeclaredMethod("deserializeOneObject", byte[].class); + } + catch (ClassNotFoundException | NoSuchMethodException e) + { + throw new RuntimeException(e); + } + } + + @SuppressWarnings("unused") // called through method invocation + public static Object deserializeOneObject(byte[] bytes) + { + try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes); + ObjectInputStream ois = new ObjectInputStream(bais);) + { + return ois.readObject(); + } + catch (IOException | ClassNotFoundException e) + { + throw new RuntimeException(e); + } + } + + private static byte[] serializeOneObject(Object object) + { + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos)) + { + oos.writeObject(object); + oos.close(); + return baos.toByteArray(); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + private static T waitOn(Future f) + { + try + { + return f.get(); + } + catch (InterruptedException e) + { + throw new RuntimeException(e); + } + catch (ExecutionException e) + { + throw new RuntimeException(e.getCause()); + } + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Listen.java b/test/distributed/org/apache/cassandra/distributed/impl/Listen.java new file mode 100644 index 0000000000..cf208a1c87 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/Listen.java @@ -0,0 +1,55 @@ +/* + * 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.impl; + +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.LockSupport; +import org.apache.cassandra.distributed.api.IListen; + +public class Listen implements IListen +{ + final Instance instance; + public Listen(Instance instance) + { + this.instance = instance; + } + + public Cancel schema(Runnable onChange) + { + final AtomicBoolean cancel = new AtomicBoolean(); + instance.isolatedExecutor.execute(() -> { + UUID prev = instance.schemaVersion(); + while (true) + { + if (cancel.get()) + return; + + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10L)); + + UUID cur = instance.schemaVersion(); + if (!prev.equals(cur)) + onChange.run(); + prev = cur; + } + }); + return () -> cancel.set(true); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/Message.java b/test/distributed/org/apache/cassandra/distributed/impl/Message.java similarity index 64% rename from test/distributed/org/apache/cassandra/distributed/Message.java rename to test/distributed/org/apache/cassandra/distributed/impl/Message.java index b5492a2a2c..6f8085c6e7 100644 --- a/test/distributed/org/apache/cassandra/distributed/Message.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Message.java @@ -16,18 +16,19 @@ * limitations under the License. */ -package org.apache.cassandra.distributed; +package org.apache.cassandra.distributed.impl; +import org.apache.cassandra.distributed.api.IMessage; import org.apache.cassandra.locator.InetAddressAndPort; // a container for simplifying the method signature for per-instance message handling/delivery -public class Message +public class Message implements IMessage { - public final int verb; - public final byte[] bytes; - public final int id; - public final int version; - public final InetAddressAndPort from; + private final int verb; + private final byte[] bytes; + private final int id; + private final int version; + private final InetAddressAndPort from; public Message(int verb, byte[] bytes, int id, int version, InetAddressAndPort from) { @@ -37,5 +38,35 @@ public class Message this.version = version; this.from = from; } + + @Override + public int verb() + { + return verb; + } + + @Override + public byte[] bytes() + { + return bytes; + } + + @Override + public int id() + { + return id; + } + + @Override + public int version() + { + return version; + } + + @Override + public InetAddressAndPort from() + { + return from; + } } diff --git a/test/distributed/org/apache/cassandra/distributed/MessageFilters.java b/test/distributed/org/apache/cassandra/distributed/impl/MessageFilters.java similarity index 78% rename from test/distributed/org/apache/cassandra/distributed/MessageFilters.java rename to test/distributed/org/apache/cassandra/distributed/impl/MessageFilters.java index 47eb52bcbd..a72c7a56b4 100644 --- a/test/distributed/org/apache/cassandra/distributed/MessageFilters.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/MessageFilters.java @@ -16,36 +16,44 @@ * limitations under the License. */ -package org.apache.cassandra.distributed; +package org.apache.cassandra.distributed.impl; import java.util.Arrays; import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; import java.util.function.BiConsumer; +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.distributed.api.IMessage; +import org.apache.cassandra.distributed.api.IMessageFilters; +import org.apache.cassandra.distributed.api.ICluster; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.net.MessagingService; -public class MessageFilters +public class MessageFilters implements IMessageFilters { - private final TestCluster cluster; + private final ICluster cluster; private final Set filters = new CopyOnWriteArraySet<>(); - public MessageFilters(TestCluster cluster) + public MessageFilters(AbstractCluster cluster) { this.cluster = cluster; } - BiConsumer filter(BiConsumer applyIfNotFiltered) + public BiConsumer filter(BiConsumer applyIfNotFiltered) { return (toAddress, message) -> { - int from = cluster.get(message.from).config.num; - int to = cluster.get(toAddress).config.num; - int verb = message.verb; + IInstance from = cluster.get(message.from()); + IInstance to = cluster.get(toAddress); + if (from == null || to == null) + return; // cannot deliver + int fromNum = from.config().num(); + int toNum = to.config().num(); + int verb = message.verb(); for (Filter filter : filters) { - if (filter.matches(from, to, verb)) + if (filter.matches(fromNum, toNum, verb)) return; } @@ -53,7 +61,7 @@ public class MessageFilters }; } - public class Filter + public class Filter implements IMessageFilters.Filter { final int[] from; final int[] to; @@ -120,7 +128,7 @@ public class MessageFilters } } - public class Builder + public class Builder implements IMessageFilters.Builder { int[] from; int[] to; @@ -154,7 +162,8 @@ public class MessageFilters } } - public Builder verbs(MessagingService.Verb ... verbs) + @Override + public Builder verbs(MessagingService.Verb... verbs) { int[] ids = new int[verbs.length]; for (int i = 0 ; i < verbs.length ; ++i) @@ -162,11 +171,13 @@ public class MessageFilters return new Builder(ids); } + @Override public Builder allVerbs() { return new Builder(null); } + @Override public void reset() { filters.clear(); diff --git a/test/distributed/org/apache/cassandra/distributed/RowUtil.java b/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java similarity index 97% rename from test/distributed/org/apache/cassandra/distributed/RowUtil.java rename to test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java index bce896d784..c3b129bac8 100644 --- a/test/distributed/org/apache/cassandra/distributed/RowUtil.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.cassandra.distributed; +package org.apache.cassandra.distributed.impl; import java.nio.ByteBuffer; import java.util.List; diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Versions.java b/test/distributed/org/apache/cassandra/distributed/impl/Versions.java new file mode 100644 index 0000000000..dba1c136f2 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/Versions.java @@ -0,0 +1,187 @@ +/* + * 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.impl; + +import java.io.File; +import java.net.MalformedURLException; +import java.net.URL; +import java.net.URLClassLoader; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.utils.FBUtilities; + +public class Versions +{ + private static final Logger logger = LoggerFactory.getLogger(Versions.class); + public static Version CURRENT = new Version(FBUtilities.getReleaseVersionString(), ((URLClassLoader)Versions.class.getClassLoader()).getURLs()); + + public enum Major + { + v22("2\\.2\\.([0-9]+)"), + v30("3\\.0\\.([0-9]+)"), + v3X("3\\.([1-9]|1[01])(\\.([0-9]+))?"), + v4("4\\.([0-9]+)"); + final Pattern pattern; + Major(String verify) + { + this.pattern = Pattern.compile(verify); + } + + static Major fromFull(String version) + { + if (version.isEmpty()) + throw new IllegalArgumentException(version); + switch (version.charAt(0)) + { + case '2': + if (version.startsWith("2.2")) + return v22; + throw new IllegalArgumentException(version); + case '3': + if (version.startsWith("3.0")) + return v30; + return v3X; + case '4': + return v4; + default: + throw new IllegalArgumentException(version); + } + } + + // verify that the version string is valid for this major version + boolean verify(String version) + { + return pattern.matcher(version).matches(); + } + + // sort two strings of the same major version as this enum instance + int compare(String a, String b) + { + Matcher ma = pattern.matcher(a); + Matcher mb = pattern.matcher(a); + if (!ma.matches()) throw new IllegalArgumentException(a); + if (!mb.matches()) throw new IllegalArgumentException(b); + int result = Integer.compare(Integer.parseInt(ma.group(1)), Integer.parseInt(mb.group(1))); + if (result == 0 && this == v3X && (ma.group(3) != null || mb.group(3) != null)) + { + if (ma.group(3) != null && mb.group(3) != null) + { + result = Integer.compare(Integer.parseInt(ma.group(3)), Integer.parseInt(mb.group(3))); + } + else + { + result = ma.group(3) != null ? 1 : -1; + } + } + // sort descending + return -result; + } + } + + public static class Version + { + public final Major major; + public final String version; + public final URL[] classpath; + + public Version(String version, URL[] classpath) + { + this(Major.fromFull(version), version, classpath); + } + public Version(Major major, String version, URL[] classpath) + { + this.major = major; + this.version = version; + this.classpath = classpath; + } + } + + private final Map> versions; + public Versions(Map> versions) + { + this.versions = versions; + } + + public Version get(String full) + { + return versions.get(Major.fromFull(full)) + .stream() + .filter(v -> full.equals(v.version)) + .findFirst() + .orElseThrow(() -> new RuntimeException("No version " + full + " found")); + } + + public Version getLatest(Major major) + { + return versions.get(major).stream().findFirst().orElseThrow(() -> new RuntimeException("No " + major + " versions found")); + } + + public static Versions find() + { + logger.info("Looking for dtest jars in " + new File("build").getAbsolutePath()); + final Pattern pattern = Pattern.compile("dtest-([0-9.]+)\\.jar"); + final Map> versions = new HashMap<>(); + for (Major major : Major.values()) + versions.put(major, new ArrayList<>()); + + for (File file : new File("build").listFiles()) + { + Matcher m = pattern.matcher(file.getName()); + if (!m.matches()) + continue; + String version = m.group(1); + Major major = Major.fromFull(version); + versions.get(major).add(new Version(major, version, new URL[] { toURL(file) })); + } + + for (Map.Entry> e : versions.entrySet()) + { + if (e.getValue().isEmpty()) + continue; + Collections.sort(e.getValue(), Comparator.comparing(v -> v.version, e.getKey()::compare)); + logger.info("Found " + e.getValue().stream().map(v -> v.version).collect(Collectors.joining(", "))); + } + + return new Versions(versions); + } + + public static URL toURL(File file) + { + try + { + return file.toURI().toURL(); + } + catch (MalformedURLException e) + { + throw new IllegalArgumentException(e); + } + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java b/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java similarity index 79% rename from test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java rename to test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java index dd23f8dd36..75131b3792 100644 --- a/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java @@ -16,19 +16,20 @@ * limitations under the License. */ -package org.apache.cassandra.distributed; +package org.apache.cassandra.distributed.test; import org.junit.Assert; import org.junit.Test; import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.distributed.Cluster; public class DistributedReadWritePathTest extends DistributedTestBase { @Test public void coordinatorRead() throws Throwable { - try (TestCluster cluster = createCluster(3)) + try (Cluster cluster = init(Cluster.create(3))) { cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); @@ -36,7 +37,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase cluster.get(2).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 2, 2)"); cluster.get(3).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 3, 3)"); - assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?", + assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?", ConsistencyLevel.ALL, 1), row(1, 1, 1), @@ -48,11 +49,11 @@ public class DistributedReadWritePathTest extends DistributedTestBase @Test public void coordinatorWrite() throws Throwable { - try (TestCluster cluster = createCluster(3)) + try (Cluster cluster = init(Cluster.create(3))) { cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); - cluster.coordinator().execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)", + cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)", ConsistencyLevel.QUORUM); for (int i = 0; i < 3; i++) @@ -61,7 +62,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase row(1, 1, 1)); } - assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", + assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.QUORUM), row(1, 1, 1)); } @@ -70,7 +71,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase @Test public void readRepairTest() throws Throwable { - try (TestCluster cluster = createCluster(3)) + try (Cluster cluster = init(Cluster.create(3))) { cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); @@ -79,7 +80,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1")); - assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", + assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.QUORUM), row(1, 1, 1)); @@ -92,7 +93,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase @Test public void writeWithSchemaDisagreement() throws Throwable { - try (TestCluster cluster = createCluster(3)) + try (Cluster cluster = init(Cluster.create(3))) { cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v1 int, PRIMARY KEY (pk, ck))"); @@ -106,7 +107,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase Exception thrown = null; try { - cluster.coordinator().execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v1, v2) VALUES (2, 2, 2, 2)", + cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v1, v2) VALUES (2, 2, 2, 2)", ConsistencyLevel.QUORUM); } catch (RuntimeException e) @@ -114,15 +115,15 @@ public class DistributedReadWritePathTest extends DistributedTestBase thrown = e; } - Assert.assertTrue(thrown.getMessage().contains("Exception occurred on the node")); - Assert.assertTrue(thrown.getCause().getMessage().contains("Unknown column v2 during deserialization")); + Assert.assertTrue(thrown.getMessage().contains("Exception occurred on node")); + Assert.assertTrue(thrown.getCause().getCause().getCause().getMessage().contains("Unknown column v2 during deserialization")); } } @Test public void readWithSchemaDisagreement() throws Throwable { - try (TestCluster cluster = createCluster(3)) + try (Cluster cluster = init(Cluster.create(3))) { cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v1 int, PRIMARY KEY (pk, ck))"); @@ -136,7 +137,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase Exception thrown = null; try { - assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", + assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.ALL), row(1, 1, 1, null)); } @@ -144,8 +145,8 @@ public class DistributedReadWritePathTest extends DistributedTestBase { thrown = e; } - Assert.assertTrue(thrown.getMessage().contains("Exception occurred on the node")); - Assert.assertTrue(thrown.getCause().getMessage().contains("Unknown column v2 during deserialization")); + Assert.assertTrue(thrown.getMessage().contains("Exception occurred on node")); + Assert.assertTrue(thrown.getCause().getCause().getCause().getMessage().contains("Unknown column v2 during deserialization")); } } } diff --git a/test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java similarity index 90% rename from test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java rename to test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java index 64ef8592ae..e442603be7 100644 --- a/test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java @@ -16,16 +16,18 @@ * limitations under the License. */ -package org.apache.cassandra.distributed; +package org.apache.cassandra.distributed.test; import java.util.Arrays; import org.junit.Assert; import org.junit.BeforeClass; +import org.apache.cassandra.distributed.impl.AbstractCluster; + public class DistributedTestBase { - static String KEYSPACE = "distributed_test_keyspace"; + public static String KEYSPACE = "distributed_test_keyspace"; @BeforeClass public static void setup() @@ -33,11 +35,9 @@ public class DistributedTestBase System.setProperty("org.apache.cassandra.disable_mbean_registration", "true"); } - TestCluster createCluster(int nodeCount) throws Throwable + protected static > C init(C cluster) { - TestCluster cluster = TestCluster.create(nodeCount); - cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + nodeCount + "};"); - + cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + cluster.size() + "};"); return cluster; } diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTest.java new file mode 100644 index 0000000000..2c7f7bc241 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTest.java @@ -0,0 +1,52 @@ +/* + * 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.upgrade; + +import org.junit.Test; + +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.distributed.impl.Versions; +import org.apache.cassandra.distributed.test.DistributedTestBase; + +public class UpgradeTest extends UpgradeTestBase +{ + + @Test + public void upgradeTest() throws Throwable + { + new TestCase() + .upgrade(Versions.Major.v22, Versions.Major.v30) + .setup((cluster) -> { + cluster.schemaChange("CREATE TABLE " + DistributedTestBase.KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + + cluster.get(1).executeInternal("INSERT INTO " + DistributedTestBase.KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)"); + cluster.get(2).executeInternal("INSERT INTO " + DistributedTestBase.KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 2, 2)"); + cluster.get(3).executeInternal("INSERT INTO " + DistributedTestBase.KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 3, 3)"); + }) + .runAfterClusterUpgrade((cluster) -> { + DistributedTestBase.assertRows(cluster.coordinator(1).execute("SELECT * FROM " + DistributedTestBase.KEYSPACE + ".tbl WHERE pk = ?", + ConsistencyLevel.ALL, + 1), + DistributedTestBase.row(1, 1, 1), + DistributedTestBase.row(1, 2, 2), + DistributedTestBase.row(1, 3, 3)); + }).run(); + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java b/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java new file mode 100644 index 0000000000..812bdbe375 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java @@ -0,0 +1,156 @@ +/* + * 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.upgrade; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.apache.cassandra.distributed.UpgradeableCluster; +import org.apache.cassandra.distributed.impl.Instance; +import org.apache.cassandra.distributed.impl.Versions; +import org.apache.cassandra.distributed.impl.Versions.Version; +import org.apache.cassandra.distributed.test.DistributedTestBase; + +import static org.apache.cassandra.distributed.impl.Versions.*; + +public class UpgradeTestBase extends DistributedTestBase +{ + public static interface RunOnCluster + { + public void run(UpgradeableCluster cluster) throws Throwable; + } + + public static interface RunOnClusterAndNode + { + public void run(UpgradeableCluster cluster, int node) throws Throwable; + } + + public static class TestVersions + { + final Version initial; + final Version[] upgrade; + + public TestVersions(Version initial, Version ... upgrade) + { + this.initial = initial; + this.upgrade = upgrade; + } + } + + public static class TestCase implements Instance.ThrowingRunnable + { + private final Versions versions; + private final List upgrade = new ArrayList<>(); + private int nodeCount = 3; + private RunOnCluster setup; + private RunOnClusterAndNode runAfterNodeUpgrade; + private RunOnCluster runAfterClusterUpgrade; + + public TestCase() + { + this(find()); + } + + public TestCase(Versions versions) + { + this.versions = versions; + } + + public TestCase nodes(int nodeCount) + { + this.nodeCount = nodeCount; + return this; + } + + public TestCase upgrade(Major initial, Major ... upgrade) + { + this.upgrade.add(new TestVersions(versions.getLatest(initial), + Arrays.stream(upgrade) + .map(versions::getLatest) + .toArray(Version[]::new))); + return this; + } + + public TestCase upgrade(Version initial, Version ... upgrade) + { + this.upgrade.add(new TestVersions(initial, upgrade)); + return this; + } + + public TestCase setup(RunOnCluster setup) + { + this.setup = setup; + return this; + } + + public TestCase runAfterNodeUpgrade(RunOnClusterAndNode runAfterNodeUpgrade) + { + this.runAfterNodeUpgrade = runAfterNodeUpgrade; + return this; + } + + public TestCase runAfterClusterUpgrade(RunOnCluster runAfterClusterUpgrade) + { + this.runAfterClusterUpgrade = runAfterClusterUpgrade; + return this; + } + + public void run() throws Throwable + { + if (setup == null) + throw new AssertionError(); + if (upgrade.isEmpty()) + throw new AssertionError(); + if (runAfterClusterUpgrade == null && runAfterNodeUpgrade == null) + throw new AssertionError(); + if (runAfterClusterUpgrade == null) + runAfterClusterUpgrade = (c) -> {}; + if (runAfterNodeUpgrade == null) + runAfterNodeUpgrade = (c, n) -> {}; + + for (TestVersions upgrade : this.upgrade) + { + try (UpgradeableCluster cluster = init(UpgradeableCluster.create(nodeCount, upgrade.initial))) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + + cluster.get(1).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)"); + cluster.get(2).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 2, 2)"); + cluster.get(3).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 3, 3)"); + + for (Version version : upgrade.upgrade) + { + for (int n = 1 ; n <= nodeCount ; ++n) + { + cluster.get(n).shutdown(); + cluster.get(n).setVersion(version); + cluster.get(n).startup(); + runAfterNodeUpgrade.run(cluster, n); + } + + runAfterClusterUpgrade.run(cluster); + } + } + + } + } + } + +} diff --git a/test/unit/org/apache/cassandra/LogbackStatusListener.java b/test/unit/org/apache/cassandra/LogbackStatusListener.java index 0ec73d9d41..d16058bf61 100644 --- a/test/unit/org/apache/cassandra/LogbackStatusListener.java +++ b/test/unit/org/apache/cassandra/LogbackStatusListener.java @@ -33,7 +33,7 @@ import ch.qos.logback.classic.LoggerContext; import ch.qos.logback.classic.spi.LoggerContextListener; import ch.qos.logback.core.status.Status; import ch.qos.logback.core.status.StatusListener; -import org.apache.cassandra.distributed.InstanceClassLoader; +import org.apache.cassandra.distributed.impl.InstanceClassLoader; /* * Listen for logback readiness and then redirect stdout/stderr to logback