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

Merge branch 'cassandra-2.2' into cassandra-3.0

This commit is contained in:
Benedict Elliott Smith 2019-02-05 10:51:28 +00:00
parent c94a6aa7e5 c100f1ac40
commit aa2a5c6994
44 changed files with 2506 additions and 1000 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.circleci/config.yml
View File

@ -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

4
CHANGES.txt
View File

@ -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)

21
build.xml
View File

@ -880,6 +880,16 @@
</manifest>
</jar>
<!-- Test Jar -->
<jar jarfile="${build.dir}/${ant.project.name}-test-${version}.jar">
<fileset dir="${test.classes}"/>
<manifest>
<attribute name="Implementation-Title" value="Cassandra"/>
<attribute name="Implementation-Version" value="${version}"/>
<attribute name="Implementation-Vendor" value="Apache"/>
</manifest>
</jar>
<!-- Clientutil Jar -->
<!-- TODO: write maven pom here -->
<jar jarfile="${build.dir}/${ant.project.name}-clientutil-${version}.jar">
@ -1159,6 +1169,15 @@
</copy>
</target>
<target name="dtest-jar" depends="build-test, build" description="Create dtest-compatible jar, including all dependencies">
<jar jarfile="${build.dir}/dtest-${base.version}.jar">
<zipgroupfileset dir="${build.lib}" includes="*.jar" excludes="META-INF/*.SF"/>
<fileset dir="${build.classes.main}"/>
<fileset dir="${test.classes}"/>
<fileset dir="${test.conf}" />
</jar>
</target>
<!-- Defines how to run a set of tests. If you change the defaults for attributes
you should also update them in testmacro.,
The two are split because the helper doesn't generate
@ -1728,7 +1747,7 @@
<target name="test" depends="build-test,get-cores,get-mem" description="Parallel Test Runner">
<path id="all-test-classes-path">
<fileset dir="${test.unit.src}" includes="**/${test.name}.java" />
<fileset dir="${test.unit.src}" includes="**/${test.name}.java" excludes="**/distributed/test/UpgradeTest*.java" />
</path>
<property name="all-test-classes" refid="all-test-classes-path"/>
<testparallel testdelegate="testlist"/>

12
src/java/org/apache/cassandra/concurrent/AbstractLocalAwareExecutorService.java
View File

@ -29,9 +29,6 @@ import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.io.FSError;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
@ -168,15 +165,10 @@ public abstract class AbstractLocalAwareExecutorService implements LocalAwareExe
}
catch (Throwable t)
{
logger.error("Uncaught exception on thread {}", Thread.currentThread(), t);
JVMStabilityInspector.inspectThrowable(t);
logger.warn("Uncaught exception on thread {}", Thread.currentThread(), t);
result = t;
failure = true;
if (t instanceof CorruptSSTableException)
FileUtils.handleCorruptSSTable((CorruptSSTableException) t);
else if (t instanceof FSError)
FileUtils.handleFSError((FSError) t);
else
JVMStabilityInspector.inspectThrowable(t);
}
finally
{

3
src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java
View File

@ -76,8 +76,9 @@ public class InfiniteLoopExecutor
thread.interrupt();
}
public void awaitTermination(long time, TimeUnit unit) throws InterruptedException
public boolean awaitTermination(long time, TimeUnit unit) throws InterruptedException
{
thread.join(unit.toMillis(time));
return !thread.isAlive();
}
}

6
src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java
View File

@ -31,6 +31,9 @@ import io.netty.util.concurrent.FastThreadLocalThread;
public class NamedThreadFactory implements ThreadFactory
{
private static volatile String globalPrefix;
public static void setGlobalPrefix(String prefix) { globalPrefix = prefix; }
public final String id;
private final int priority;
private final ClassLoader contextClassLoader;
@ -58,7 +61,8 @@ public class NamedThreadFactory implements ThreadFactory
public Thread newThread(Runnable runnable)
{
String name = id + ':' + n.getAndIncrement();
Thread thread = new FastThreadLocalThread(threadGroup, threadLocalDeallocator(runnable), name);
String prefix = globalPrefix;
Thread thread = new FastThreadLocalThread(threadGroup, threadLocalDeallocator(runnable), prefix != null ? prefix + name : name);
thread.setPriority(priority);
thread.setDaemon(true);
if (contextClassLoader != null)

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

@ -128,10 +128,6 @@ public class DatabaseDescriptor
{
conf = new Config();
}
else if (Config.getOverrideLoadConfig() != null)
{
applyConfig(Config.getOverrideLoadConfig().get());
}
else
{
applyConfig(loadConfig());
@ -143,9 +139,11 @@ public class DatabaseDescriptor
}
}
@VisibleForTesting
public static Config loadConfig() throws ConfigurationException
{
if (Config.getOverrideLoadConfig() != null)
return Config.getOverrideLoadConfig().get();
String loaderClass = System.getProperty("cassandra.config.loader");
ConfigurationLoader loader = loaderClass == null
? new YamlConfigurationLoader()

9
src/java/org/apache/cassandra/utils/FBUtilities.java
View File

@ -61,6 +61,7 @@ import org.apache.cassandra.schema.CompressionParams;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputBufferFixed;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.AsyncOneResponse;
import org.codehaus.jackson.JsonFactory;
@ -158,6 +159,14 @@ public class FBUtilities
return broadcastInetAddress;
}
/**
* <b>THIS IS FOR TESTING ONLY!!</b>
*/
@VisibleForTesting
public static void setBroadcastInetAddress(InetAddress addr)
{
broadcastInetAddress = addr;
}
public static InetAddress getBroadcastRpcAddress()
{

2
test/conf/logback-dtest.xml
View File

@ -18,7 +18,7 @@
-->
<configuration debug="false" scan="true" scanPeriod="60 seconds">
<define name="instance_id" class="org.apache.cassandra.distributed.InstanceIDDefiner" />
<define name="instance_id" class="org.apache.cassandra.distributed.impl.InstanceIDDefiner" />
<!-- Shutdown hook ensures that async appender flushes -->
<shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/>

57
test/distributed/org/apache/cassandra/distributed/Cluster.java Normal file
View File

@ -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<IInvokableInstance> implements ICluster, AutoCloseable
{
private Cluster(File root, Versions.Version version, List<InstanceConfig> 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);
}
}

78
test/distributed/org/apache/cassandra/distributed/Coordinator.java
View File

@ -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<ByteBuffer> 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);
}
}

109
test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java
View File

@ -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<String> isCommonClassName;
InstanceClassLoader(int id, URL[] urls, Predicate<String> 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<ClassLoader> createFactory(URLClassLoader contextClassLoader)
{
Set<String> 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());
}
}

87
test/distributed/org/apache/cassandra/distributed/InstanceConfig.java
View File

@ -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);
}
}

213
test/distributed/org/apache/cassandra/distributed/InvokableInstance.java
View File

@ -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<T> extends Callable<T> { public T call(); }
public interface SerializableCallable<T> extends CallableNoExcept<T>, Serializable { }
public <T> CallableNoExcept<T> callsOnInstance(SerializableCallable<T> call) { return invokesOnExecutor((SerializableCallable<T>) transferOneObject(call), isolatedExecutor); }
public <T> T callOnInstance(SerializableCallable<T> 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<T> extends Consumer<T>, Serializable {}
public <T> Consumer<T> acceptsOnInstance(SerializableConsumer<T> consumer) { return invokesOnExecutor((SerializableConsumer<T>) transferOneObject(consumer), isolatedExecutor); }
public interface SerializableBiConsumer<T1, T2> extends BiConsumer<T1, T2>, Serializable {}
public <T1, T2> BiConsumer<T1, T2> acceptsOnInstance(SerializableBiConsumer<T1, T2> consumer) { return invokesOnExecutor((SerializableBiConsumer<T1, T2>) transferOneObject(consumer), isolatedExecutor); }
public interface SerializableFunction<I, O> extends Function<I, O>, Serializable {}
public <I, O> Function<I, O> appliesOnInstance(SerializableFunction<I, O> f) { return invokesOnExecutor((SerializableFunction<I, O>) transferOneObject(f), isolatedExecutor); }
public interface SerializableBiFunction<I1, I2, O> extends BiFunction<I1, I2, O>, Serializable {}
public <I1, I2, O> BiFunction<I1, I2, O> appliesOnInstance(SerializableBiFunction<I1, I2, O> f) { return invokesOnExecutor((SerializableBiFunction<I1, I2, O>) transferOneObject(f), isolatedExecutor); }
public interface TriFunction<I1, I2, I3, O>
{
O apply(I1 i1, I2 i2, I3 i3);
}
public interface SerializableTriFunction<I1, I2, I3, O> extends Serializable, TriFunction<I1, I2, I3, O> { }
public <I1, I2, I3, O> TriFunction<I1, I2, I3, O> appliesOnInstance(SerializableTriFunction<I1, I2, I3, O> f) { return invokesOnExecutor((SerializableTriFunction<I1, I2, I3, O>) transferOneObject(f), isolatedExecutor); }
public interface InstanceFunction<I, O> extends SerializableBiFunction<Instance, I, O> {}
// E must be a functional interface, and lambda must be implemented by a lambda function
public <E extends Serializable> 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 <V> CallableNoExcept<V> invokesOnExecutor(SerializableCallable<V> 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 <A> Consumer<A> invokesOnExecutor(SerializableConsumer<A> consumer, ExecutorService invokeOn)
{
return (a) -> invokesOnExecutor(() -> consumer.accept(a), invokeOn).run();
}
private static <A, B> BiConsumer<A, B> invokesOnExecutor(SerializableBiConsumer<A, B> consumer, ExecutorService invokeOn)
{
return (a, b) -> invokesOnExecutor(() -> consumer.accept(a, b), invokeOn).run();
}
private static <A, B> Function<A, B> invokesOnExecutor(SerializableFunction<A, B> f, ExecutorService invokeOn)
{
return (a) -> invokesOnExecutor(() -> f.apply(a), invokeOn).call();
}
private static <A, B, C> BiFunction<A, B, C> invokesOnExecutor(SerializableBiFunction<A, B, C> f, ExecutorService invokeOn)
{
return (a, b) -> invokesOnExecutor(() -> f.apply(a, b), invokeOn).call();
}
private static <A, B, C, D> SerializableTriFunction<A, B, C, D> invokesOnExecutor(SerializableTriFunction<A, B, C, D> f, ExecutorService invokeOn)
{
return (a, b, c) -> invokesOnExecutor(() -> f.apply(a, b, c), invokeOn).call();
}
void shutdown()
{
isolatedExecutor.shutdownNow();
}
}

280
test/distributed/org/apache/cassandra/distributed/TestCluster.java
View File

@ -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<Instance> instances;
private final Coordinator coordinator;
private final Map<InetAddressAndPort, Instance> instanceMap;
private final MessageFilters filters;
private TestCluster(File root, List<Instance> 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<Instance> 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<ClassLoader> classLoaderFactory = InstanceClassLoader.createFactory(
(URLClassLoader) Thread.currentThread().getContextClassLoader());
List<Instance> 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<Future<?>> 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<Future<?>> futures)
{
FBUtilities.waitOnFutures(futures);
Set<Thread> 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));
}
}
}

70
test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java Normal file
View File

@ -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<IUpgradeableInstance> implements ICluster, AutoCloseable
{
private UpgradeableCluster(File root, Versions.Version version, List<InstanceConfig> 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);
}
}

34
test/distributed/org/apache/cassandra/distributed/api/ICluster.java Normal file
View File

@ -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<? extends IInstance> stream();
IMessageFilters filters();
}

27
test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java Normal file
View File

@ -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);
}

44
test/distributed/org/apache/cassandra/distributed/api/IInstance.java Normal file
View File

@ -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);
}

41
test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java Normal file
View File

@ -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);
}

127
test/distributed/org/apache/cassandra/distributed/api/IIsolatedExecutor.java Normal file
View File

@ -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<O> extends Callable<O> { public O call(); }
public interface SerializableCallable<O> extends CallableNoExcept<O>, Serializable { }
public interface SerializableRunnable extends Runnable, Serializable {}
public interface SerializableConsumer<O> extends Consumer<O>, Serializable {}
public interface SerializableBiConsumer<I1, I2> extends BiConsumer<I1, I2>, Serializable {}
public interface SerializableFunction<I, O> extends Function<I, O>, Serializable {}
public interface SerializableBiFunction<I1, I2, O> extends BiFunction<I1, I2, O>, Serializable {}
public interface TriFunction<I1, I2, I3, O>
{
O apply(I1 i1, I2 i2, I3 i3);
}
public interface SerializableTriFunction<I1, I2, I3, O> extends Serializable, TriFunction<I1, I2, I3, O> { }
void shutdown();
/**
* Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
*/
<O> CallableNoExcept<Future<O>> async(CallableNoExcept<O> call);
/**
* Convert the execution to one performed synchronously on the IsolatedExecutor
*/
<O> CallableNoExcept<O> sync(CallableNoExcept<O> call);
/**
* Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
*/
CallableNoExcept<Future<?>> 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
*/
<I> Function<I, Future<?>> async(Consumer<I> consumer);
/**
* Convert the execution to one performed synchronously on the IsolatedExecutor
*/
<I> Consumer<I> sync(Consumer<I> consumer);
/**
* Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
*/
<I1, I2> BiFunction<I1, I2, Future<?>> async(BiConsumer<I1, I2> consumer);
/**
* Convert the execution to one performed synchronously on the IsolatedExecutor
*/
<I1, I2> BiConsumer<I1, I2> sync(BiConsumer<I1, I2> consumer);
/**
* Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
*/
<I, O> Function<I, Future<O>> async(Function<I, O> f);
/**
* Convert the execution to one performed synchronously on the IsolatedExecutor
*/
<I, O> Function<I, O> sync(Function<I, O> f);
/**
* Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
*/
<I1, I2, O> BiFunction<I1, I2, Future<O>> async(BiFunction<I1, I2, O> f);
/**
* Convert the execution to one performed synchronously on the IsolatedExecutor
*/
<I1, I2, O> BiFunction<I1, I2, O> sync(BiFunction<I1, I2, O> f);
/**
* Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
*/
<I1, I2, I3, O> TriFunction<I1, I2, I3, Future<O>> async(TriFunction<I1, I2, I3, O> f);
/**
* Convert the execution to one performed synchronously on the IsolatedExecutor
*/
<I1, I2, I3, O> TriFunction<I1, I2, I3, O> sync(TriFunction<I1, I2, I3, O> f);
}

26
test/distributed/org/apache/cassandra/distributed/api/IListen.java Normal file
View File

@ -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);
}

33
test/distributed/org/apache/cassandra/distributed/api/IMessage.java Normal file
View File

@ -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();
}

30
test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java → test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java
View File

@ -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<InetAddressAndPort, IMessage> filter(BiConsumer<InetAddressAndPort, IMessage> applyIfNotFiltered);
}

406
test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java Normal file
View File

@ -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<I extends IInstance> 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<I> instances;
private final Map<InetAddressAndPort, I> 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<IInstanceConfig, ClassLoader, Instance>)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<InstanceConfig> 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<I> stream() { return instances.stream(); }
public void forEach(IIsolatedExecutor.SerializableRunnable runnable) { forEach(i -> i.sync(runnable)); }
public void forEach(Consumer<? super I> consumer) { instances.forEach(consumer); }
public void parallelForEach(IIsolatedExecutor.SerializableConsumer<? super I> 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<IListen.Cancel> 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<I extends IInstance, C extends AbstractCluster<I>>
{
C newCluster(File root, Versions.Version version, List<InstanceConfig> configs, ClassLoader sharedClassLoader);
}
protected static <I extends IInstance, C extends AbstractCluster<I>> C
create(int nodeCount, Factory<I, C> factory) throws Throwable
{
return create(nodeCount, Files.createTempDirectory("dtests").toFile(), factory);
}
protected static <I extends IInstance, C extends AbstractCluster<I>> C
create(int nodeCount, File root, Factory<I, C> factory)
{
return create(nodeCount, Versions.CURRENT, root, factory);
}
protected static <I extends IInstance, C extends AbstractCluster<I>> C
create(int nodeCount, Versions.Version version, Factory<I, C> factory) throws IOException
{
return create(nodeCount, version, Files.createTempDirectory("dtests").toFile(), factory);
}
protected static <I extends IInstance, C extends AbstractCluster<I>> C
create(int nodeCount, Versions.Version version, File root, Factory<I, C> factory)
{
root.mkdirs();
setupLogging(root);
ClassLoader sharedClassLoader = Thread.currentThread().getContextClassLoader();
List<InstanceConfig> 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<Future<?>> futures)
{
FBUtilities.waitOnFutures(futures);
Set<Thread> 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));
}
}
}

75
test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java Normal file
View File

@ -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<ByteBuffer> 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();
}
}

196
test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java Normal file
View File

@ -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 extends Serializable> 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 <O> CallableNoExcept<Future<O>> async(CallableNoExcept<O> call)
{
return delegate().async(call);
}
@Override
public <O> CallableNoExcept<O> sync(CallableNoExcept<O> call)
{
return delegate().sync(call);
}
@Override
public CallableNoExcept<Future<?>> async(Runnable run)
{
return delegate().async(run);
}
@Override
public Runnable sync(Runnable run)
{
return delegate().sync(run);
}
@Override
public <I> Function<I, Future<?>> async(Consumer<I> consumer)
{
return delegate().async(consumer);
}
@Override
public <I> Consumer<I> sync(Consumer<I> consumer)
{
return delegate().sync(consumer);
}
@Override
public <I1, I2> BiFunction<I1, I2, Future<?>> async(BiConsumer<I1, I2> consumer)
{
return delegate().async(consumer);
}
@Override
public <I1, I2> BiConsumer<I1, I2> sync(BiConsumer<I1, I2> consumer)
{
return delegate().sync(consumer);
}
@Override
public <I, O> Function<I, Future<O>> async(Function<I, O> f)
{
return delegate().async(f);
}
@Override
public <I, O> Function<I, O> sync(Function<I, O> f)
{
return delegate().sync(f);
}
@Override
public <I1, I2, O> BiFunction<I1, I2, Future<O>> async(BiFunction<I1, I2, O> f)
{
return delegate().async(f);
}
@Override
public <I1, I2, O> BiFunction<I1, I2, O> sync(BiFunction<I1, I2, O> f)
{
return delegate().sync(f);
}
@Override
public <I1, I2, I3, O> TriFunction<I1, I2, I3, Future<O>> async(TriFunction<I1, I2, I3, O> f)
{
return delegate().async(f);
}
@Override
public <I1, I2, I3, O> TriFunction<I1, I2, I3, O> sync(TriFunction<I1, I2, I3, O> f)
{
return delegate().sync(f);
}
}

68
test/distributed/org/apache/cassandra/distributed/impl/IInvokableInstance.java Normal file
View File

@ -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 <O> CallableNoExcept<Future<O>> asyncCallsOnInstance(SerializableCallable<O> call) { return async(transfer(call)); }
public default <O> CallableNoExcept<O> callsOnInstance(SerializableCallable<O> call) { return sync(transfer(call)); }
public default <O> O callOnInstance(SerializableCallable<O> call) { return callsOnInstance(call).call(); }
public default CallableNoExcept<Future<?>> 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 <I> Function<I, Future<?>> asyncAcceptsOnInstance(SerializableConsumer<I> consumer) { return async(transfer(consumer)); }
public default <I> Consumer<I> acceptsOnInstance(SerializableConsumer<I> consumer) { return sync(transfer(consumer)); }
public default <I1, I2> BiFunction<I1, I2, Future<?>> asyncAcceptsOnInstance(SerializableBiConsumer<I1, I2> consumer) { return async(transfer(consumer)); }
public default <I1, I2> BiConsumer<I1, I2> acceptsOnInstance(SerializableBiConsumer<I1, I2> consumer) { return sync(transfer(consumer)); }
public default <I, O> Function<I, Future<O>> asyncAppliesOnInstance(SerializableFunction<I, O> f) { return async(transfer(f)); }
public default <I, O> Function<I, O> appliesOnInstance(SerializableFunction<I, O> f) { return sync(transfer(f)); }
public default <I1, I2, O> BiFunction<I1, I2, Future<O>> asyncAppliesOnInstance(SerializableBiFunction<I1, I2, O> f) { return async(transfer(f)); }
public default <I1, I2, O> BiFunction<I1, I2, O> appliesOnInstance(SerializableBiFunction<I1, I2, O> f) { return sync(transfer(f)); }
public default <I1, I2, I3, O> TriFunction<I1, I2, I3, Future<O>> asyncAppliesOnInstance(SerializableTriFunction<I1, I2, I3, O> f) { return async(transfer(f)); }
public default <I1, I2, I3, O> TriFunction<I1, I2, I3, O> appliesOnInstance(SerializableTriFunction<I1, I2, I3, O> f) { return sync(transfer(f)); }
public <E extends Serializable> E transfer(E object);
}

28
test/distributed/org/apache/cassandra/distributed/impl/IUpgradeableInstance.java Normal file
View File

@ -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);
}

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

@ -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<InetAddressAndPort, Message> deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message);
BiConsumer<InetAddressAndPort, Message> deliverToInstanceIfNotFiltered = cluster.filters().filter(deliverToInstance);
BiConsumer<InetAddressAndPort, IMessage> deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message);
BiConsumer<InetAddressAndPort, IMessage> deliverToInstanceIfNotFiltered = cluster.filters().filter(deliverToInstance);
Map<InetAddress, InetAddressAndPort> 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<InetAddressAndPort, Message> 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<InetAddressAndPort, Message> deliver;
private final BiConsumer<InetAddressAndPort, IMessage> deliver;
private final Function<InetAddress, InetAddressAndPort> lookupAddressAndPort;
MessageDeliverySink(BiConsumer<InetAddressAndPort, Message> deliver, Function<InetAddress, InetAddressAndPort> lookupAddressAndPort)
MessageDeliverySink(BiConsumer<InetAddressAndPort, IMessage> deliver, Function<InetAddress, InetAddressAndPort> 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<String> initialTokens = new ArrayList<>();
List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> 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<Token> tokens = new ArrayList<>();
for (String token : initialTokens)
tokens.add(partitioner.getTokenFactory().fromString(token));
IPartitioner partitioner = FBUtilities.newPartitioner(partitionerName);
StorageService storageService = StorageService.instance;
List<Token> 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)

103
test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java Normal file
View File

@ -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<String> sharedClassNames = Arrays.stream(new Class[]
{
Pair.class,
InetAddressAndPort.class,
ParameterizedClass.class,
IInvokableInstance.class
})
.map(Class::getName)
.collect(Collectors.toSet());
private static final Predicate<String> 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<String> 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());
}
}

233
test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java Normal file
View File

@ -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<String, Object> 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<String, Object> 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();
}
}

15
test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java → test/distributed/org/apache/cassandra/distributed/impl/InstanceIDDefiner.java
View File

@ -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 = "<main>";
public static void setInstanceId(int id)
{
instanceId = "node" + id;
NamedThreadFactory.setGlobalPrefix("node" + id + "_");
}
public String getPropertyValue()
{
if (instanceId == -1)
return "<main>";
else
return "INSTANCE" + instanceId;
return instanceId;
}
}

165
test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java Normal file
View File

@ -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 <O> CallableNoExcept<Future<O>> async(CallableNoExcept<O> call) { return () -> isolatedExecutor.submit(call); }
public <O> CallableNoExcept<O> sync(CallableNoExcept<O> call) { return () -> waitOn(async(call).call()); }
public CallableNoExcept<Future<?>> async(Runnable run) { return () -> isolatedExecutor.submit(run); }
public Runnable sync(Runnable run) { return () -> waitOn(async(run).call()); }
public <I> Function<I, Future<?>> async(Consumer<I> consumer) { return (a) -> isolatedExecutor.submit(() -> consumer.accept(a)); }
public <I> Consumer<I> sync(Consumer<I> consumer) { return (a) -> waitOn(async(consumer).apply(a)); }
public <I1, I2> BiFunction<I1, I2, Future<?>> async(BiConsumer<I1, I2> consumer) { return (a, b) -> isolatedExecutor.submit(() -> consumer.accept(a, b)); }
public <I1, I2> BiConsumer<I1, I2> sync(BiConsumer<I1, I2> consumer) { return (a, b) -> waitOn(async(consumer).apply(a, b)); }
public <I, O> Function<I, Future<O>> async(Function<I, O> f) { return (a) -> isolatedExecutor.submit(() -> f.apply(a)); }
public <I, O> Function<I, O> sync(Function<I, O> f) { return (a) -> waitOn(async(f).apply(a)); }
public <I1, I2, O> BiFunction<I1, I2, Future<O>> async(BiFunction<I1, I2, O> f) { return (a, b) -> isolatedExecutor.submit(() -> f.apply(a, b)); }
public <I1, I2, O> BiFunction<I1, I2, O> sync(BiFunction<I1, I2, O> f) { return (a, b) -> waitOn(async(f).apply(a, b)); }
public <I1, I2, I3, O> TriFunction<I1, I2, I3, Future<O>> async(TriFunction<I1, I2, I3, O> f) { return (a, b, c) -> isolatedExecutor.submit(() -> f.apply(a, b, c)); }
public <I1, I2, I3, O> TriFunction<I1, I2, I3, O> sync(TriFunction<I1, I2, I3, O> f) { return (a, b, c) -> waitOn(async(f).apply(a, b, c)); }
public <E extends Serializable> E transfer(E object)
{
return (E) transferOneObject(object, classLoader, deserializeOnInstance);
}
static <E extends Serializable> E transferAdhoc(E object, ClassLoader classLoader)
{
return transferOneObject(object, classLoader, lookupDeserializeOneObject(classLoader));
}
private static <E extends Serializable> 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> T waitOn(Future<T> f)
{
try
{
return f.get();
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e.getCause());
}
}
}

55
test/distributed/org/apache/cassandra/distributed/impl/Listen.java Normal file
View File

@ -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);
}
}

45
test/distributed/org/apache/cassandra/distributed/Message.java → test/distributed/org/apache/cassandra/distributed/impl/Message.java
View File

@ -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;
}
}

35
test/distributed/org/apache/cassandra/distributed/MessageFilters.java → test/distributed/org/apache/cassandra/distributed/impl/MessageFilters.java
View File

@ -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<Filter> filters = new CopyOnWriteArraySet<>();
public MessageFilters(TestCluster cluster)
public MessageFilters(AbstractCluster cluster)
{
this.cluster = cluster;
}
BiConsumer<InetAddressAndPort, Message> filter(BiConsumer<InetAddressAndPort, Message> applyIfNotFiltered)
public BiConsumer<InetAddressAndPort, IMessage> filter(BiConsumer<InetAddressAndPort, IMessage> 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();

2
test/distributed/org/apache/cassandra/distributed/RowUtil.java → test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java
View File

@ -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;

187
test/distributed/org/apache/cassandra/distributed/impl/Versions.java Normal file
View File

@ -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<Major, List<Version>> versions;
public Versions(Map<Major, List<Version>> 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<Major, List<Version>> 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<Major, List<Version>> 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);
}
}
}

33
test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java → test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java
View File

@ -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"));
}
}
}

12
test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java → test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java
View File

@ -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 extends AbstractCluster<?>> 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;
}

52
test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTest.java Normal file
View File

@ -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();
}
}

156
test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java Normal file
View File

@ -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<TestVersions> 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);
}
}
}
}
}
}

2
test/unit/org/apache/cassandra/LogbackStatusListener.java
View File

@ -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