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

Introduce in-jvm distributed tests 

Patch by Alex Petrov and Benedict Elliott Smith; reviewed by Benedict Elliott Smith and Dinesh Joshi for CASSANDRA-14821.

Co-authored-by: Benedict Elliott Smith <benedict@apache.org>
This commit is contained in:
Alex Petrov 2018-11-16 19:41:58 +01:00 committed by Benedict Elliott Smith
parent 1d06b6686f
commit 1d5e0301f3
54 changed files with 2437 additions and 329 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.circleci/config.yml
View File

@ -163,12 +163,20 @@ 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
# 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
set -eo pipefail && cat /tmp/java_tests_${CIRCLE_NODE_INDEX}.txt | cut -c 37-1000000 | grep "Test\.java$" > /tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt
echo "** /tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt"
cat /tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt
set -eo pipefail && circleci tests split --split-by=timings --timings-type=filename --index=${CIRCLE_NODE_INDEX} --total=${CIRCLE_NODE_TOTAL} /tmp/all_java_distributed_tests.txt > /tmp/java_dtests_${CIRCLE_NODE_INDEX}.txt
set +eo pipefail && cat /tmp/java_dtests_${CIRCLE_NODE_INDEX}.txt | cut -c 44-1000000 | grep "Test\.java$" > /tmp/java_dtests_${CIRCLE_NODE_INDEX}_final.txt
echo "** /tmp/java_dtests_${CIRCLE_NODE_INDEX}_final.txt"
cat /tmp/java_dtests_${CIRCLE_NODE_INDEX}_final.txt
- run:
name: Run Unit Tests
command: |
@ -180,7 +188,11 @@ jobs:
time mv ~/cassandra /tmp
cd /tmp/cassandra
ant testclasslist -Dtest.classlistfile=/tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt
ant testclasslist -Dtest.classlistfile=/tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt -Dtest.classlistprefix=unit
if [ -s "/tmp/java_dtests_${CIRCLE_NODE_INDEX}_final.txt" ]; then
ant testclasslist -Dtest.classlistfile=/tmp/java_dtests_${CIRCLE_NODE_INDEX}_final.txt -Dtest.classlistprefix=distributed
fi
no_output_timeout: 15m
- store_test_results:
path: /tmp/cassandra/build/test/output/

18
build.xml
View File

@ -57,6 +57,7 @@
<property name="test.data" value="${test.dir}/data"/>
<property name="test.name" value="*Test"/>
<property name="test.classlistfile" value="testlist.txt"/>
<property name="test.classlistprefix" value="unit"/>
<property name="benchmark.name" value=""/>
<property name="test.methods" value=""/>
<property name="test.runners" value="1"/>
@ -65,11 +66,14 @@
<property name="test.burn.src" value="${test.dir}/burn"/>
<property name="test.microbench.src" value="${test.dir}/microbench"/>
<property name="test.pig.src" value="${test.dir}/pig"/>
<property name="test.distributed.src" value="${test.dir}/distributed"/>
<property name="dist.dir" value="${build.dir}/dist"/>
<property name="tmp.dir" value="${java.io.tmpdir}"/>
<property name="source.version" value="1.7"/>
<property name="source.test.version" value="1.8"/>
<property name="target.version" value="1.7"/>
<property name="target.test.version" value="1.8"/>
<condition property="version" value="${base.version}">
<isset property="release"/>
@ -97,7 +101,8 @@
<property name="test.timeout" value="240000" />
<property name="test.long.timeout" value="600000" />
<property name="test.burn.timeout" value="600000" />
<property name="test.burn.timeout" value="60000000" />
<property name="test.distributed.timeout" value="600000" />
<!-- default for cql tests. Can be override by -Dcassandra.test.use_prepared=false -->
<property name="cassandra.test.use_prepared" value="true" />
@ -1184,8 +1189,8 @@
encoding="utf-8"
destdir="${test.classes}"
includeantruntime="true"
source="${source.version}"
target="${target.version}">
source="${source.test.version}"
target="${target.test.version}">
<classpath>
<path refid="cassandra.classpath"/>
</classpath>
@ -1195,6 +1200,7 @@
<src path="${test.burn.src}"/>
<src path="${test.pig.src}"/>
<src path="${test.microbench.src}"/>
<src path="${test.distributed.src}"/>
</javac>
<!-- Non-java resources needed by the test suite -->
@ -1333,7 +1339,7 @@
<attribute name="test.file.list"/>
<attribute name="testlist.offset"/>
<sequential>
<testmacrohelper inputdir="${test.unit.src}" filelist="@{test.file.list}" poffset="@{testlist.offset}" exclude="**/*.java" timeout="${test.timeout}">
<testmacrohelper inputdir="${test.dir}/${test.classlistprefix}" filelist="@{test.file.list}" poffset="@{testlist.offset}" exclude="**/*.java" timeout="${test.timeout}">
<jvmarg value="-Dlegacy-sstable-root=${test.data}/legacy-sstables"/>
<jvmarg value="-Dinvalid-legacy-sstable-root=${test.data}/invalid-legacy-sstables"/>
<jvmarg value="-Dcorrupt-sstable-root=${test.data}/corrupt-sstables"/>
@ -1405,6 +1411,7 @@
</concat>
<path id="all-test-classes-path">
<fileset dir="${test.unit.src}" excludes="**/pig/*.java" includes="**/${test.name}.java" />
<fileset dir="${test.distributed.src}" includes="**/${test.name}.java" />
</path>
<property name="all-test-classes" refid="all-test-classes-path"/>
<testparallel testdelegate="testlist-compression" />
@ -1767,7 +1774,7 @@
e.g. org/apache/cassandra/hints/HintMessageTest.java -->
<target name="testclasslist" depends="build-test" description="Parallel-run tests given in file -Dtest.classlistfile (one-class-per-line, e.g. org/apache/cassandra/db/SomeTest.java)">
<path id="all-test-classes-path">
<fileset dir="${test.unit.src}" includesfile="${test.classlistfile}"/>
<fileset dir="${test.dir}/${test.classlistprefix}" includesfile="${test.classlistfile}"/>
</path>
<property name="all-test-classes" refid="all-test-classes-path"/>
<testparallel testdelegate="testlist"/>
@ -1837,6 +1844,7 @@
<classpathentry kind="src" output="build/test/classes" path="test/unit"/>
<classpathentry kind="src" output="build/test/classes" path="test/long"/>
<classpathentry kind="src" output="build/test/classes" path="test/pig"/>
<classpathentry kind="src" output="build/test/classes" path="test/distributed"/>
<classpathentry kind="src" output="build/test/classes" path="test/resources" />
<classpathentry kind="src" path="tools/stress/src"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>

1
ide/idea-iml-file.xml
View File

@ -33,6 +33,7 @@
<sourceFolder url="file://$MODULE_DIR$/test/microbench" isTestSource="true" />
<sourceFolder url="file://$MODULE_DIR$/test/burn" isTestSource="true" />
<sourceFolder url="file://$MODULE_DIR$/test/pig" isTestSource="true" />
<sourceFolder url="file://$MODULE_DIR$/test/distributed" isTestSource="true" />
<sourceFolder url="file://$MODULE_DIR$/test/resources" type="java-test-resource" />
<excludeFolder url="file://$MODULE_DIR$/.idea" />
<excludeFolder url="file://$MODULE_DIR$/.settings" />

14
src/java/org/apache/cassandra/auth/PermissionsCache.java
View File

@ -31,11 +31,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.Pair;
import javax.management.MBeanServer;
import javax.management.ObjectName;
public class PermissionsCache implements PermissionsCacheMBean
{
private static final Logger logger = LoggerFactory.getLogger(PermissionsCache.class);
@ -51,15 +49,7 @@ public class PermissionsCache implements PermissionsCacheMBean
{
this.authorizer = authorizer;
this.cache = initCache(null);
try
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME);
}
public Set<Permission> getPermissions(AuthenticatedUser user, IResource resource)

11
src/java/org/apache/cassandra/auth/RolesCache.java
View File

@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.utils.MBeanWrapper;
import javax.management.MBeanServer;
import javax.management.ObjectName;
@ -49,15 +50,7 @@ public class RolesCache implements RolesCacheMBean
{
this.roleManager = roleManager;
this.cache = initCache(null);
try
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME);
}
public Set<RoleResource> getRoles(RoleResource role)

89
src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java Normal file
View File

@ -0,0 +1,89 @@
/*
* 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.concurrent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
public class InfiniteLoopExecutor
{
private static final Logger logger = LoggerFactory.getLogger(InfiniteLoopExecutor.class);
public interface InterruptibleRunnable
{
void run() throws InterruptedException;
}
private final Thread thread;
private final InterruptibleRunnable runnable;
private volatile boolean isShutdown = false;
public InfiniteLoopExecutor(String name, InterruptibleRunnable runnable)
{
this.runnable = runnable;
this.thread = new Thread(new Runnable()
{
public void run()
{
loop();
}
}, name);
this.thread.setDaemon(true);
}
private void loop()
{
while (!isShutdown)
{
try
{
runnable.run();
}
catch (InterruptedException ie)
{
if (isShutdown)
return;
logger.error("Interrupted while executing {}, but not shutdown; continuing with loop", runnable, ie);
}
catch (Throwable t)
{
logger.error("Exception thrown by runnable, continuing with loop", t);
}
}
}
public InfiniteLoopExecutor start()
{
thread.start();
return this;
}
public void shutdown()
{
isShutdown = true;
thread.interrupt();
}
public void awaitTermination(long time, TimeUnit unit) throws InterruptedException
{
thread.join(unit.toMillis(time));
}
}

24
src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java
View File

@ -17,15 +17,13 @@
*/
package org.apache.cassandra.concurrent;
import java.lang.management.ManagementFactory;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.cassandra.metrics.ThreadPoolMetrics;
import org.apache.cassandra.utils.MBeanWrapper;
/**
* This is a wrapper class for the <i>ScheduledThreadPoolExecutor</i>. It provides an implementation
@ -75,17 +73,8 @@ public class JMXEnabledThreadPoolExecutor extends DebuggableThreadPoolExecutor i
super.prestartAllCoreThreads();
metrics = new ThreadPoolMetrics(this, jmxPath, threadFactory.id);
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
mbeanName = "org.apache.cassandra." + jmxPath + ":type=" + threadFactory.id;
try
{
mbs.registerMBean(this, new ObjectName(mbeanName));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(this, mbeanName);
}
public JMXEnabledThreadPoolExecutor(Stage stage)
@ -95,14 +84,7 @@ public class JMXEnabledThreadPoolExecutor extends DebuggableThreadPoolExecutor i
private void unregisterMBean()
{
try
{
ManagementFactory.getPlatformMBeanServer().unregisterMBean(new ObjectName(mbeanName));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.unregisterMBean(mbeanName);
// release metrics
metrics.release();

15
src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java
View File

@ -17,6 +17,11 @@
*/
package org.apache.cassandra.concurrent;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import com.google.common.annotations.VisibleForTesting;
/**
* Centralized location for shared executors
*/
@ -36,4 +41,14 @@ public class ScheduledExecutors
* This executor is used for tasks that do not need to be waited for on shutdown/drain.
*/
public static final DebuggableScheduledThreadPoolExecutor optionalTasks = new DebuggableScheduledThreadPoolExecutor("OptionalTasks");
@VisibleForTesting
public static void shutdownAndWait() throws InterruptedException
{
ExecutorService[] executors = new ExecutorService[] { scheduledTasks, nonPeriodicTasks, optionalTasks };
for (ExecutorService executor : executors)
executor.shutdown();
for (ExecutorService executor : executors)
executor.awaitTermination(60, TimeUnit.SECONDS);
}
}

15
src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java
View File

@ -21,9 +21,12 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.annotations.VisibleForTesting;
import static org.apache.cassandra.concurrent.SEPWorker.Work;
/**
@ -61,7 +64,7 @@ public class SharedExecutorPool
final AtomicLong workerId = new AtomicLong();
// the collection of executors serviced by this pool; periodically ordered by traffic volume
final List<SEPExecutor> executors = new CopyOnWriteArrayList<>();
public final List<SEPExecutor> executors = new CopyOnWriteArrayList<>();
// the number of workers currently in a spinning state
final AtomicInteger spinningCount = new AtomicInteger();
@ -109,4 +112,14 @@ public class SharedExecutorPool
executors.add(executor);
return executor;
}
@VisibleForTesting
public static void shutdownSharedPool() throws InterruptedException
{
for (SEPExecutor executor : SHARED.executors)
executor.shutdown();
for (SEPExecutor executor : SHARED.executors)
executor.awaitTermination(60, TimeUnit.SECONDS);
}
}

10
src/java/org/apache/cassandra/concurrent/StageManager.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.concurrent;
import java.util.EnumMap;
import java.util.concurrent.*;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -119,6 +120,15 @@ public class StageManager
}
};
@VisibleForTesting
public static void shutdownAndWait() throws InterruptedException
{
for (Stage stage : Stage.values())
StageManager.stages.get(stage).shutdown();
for (Stage stage : Stage.values())
StageManager.stages.get(stage).awaitTermination(60, TimeUnit.SECONDS);
}
/**
* A TPE that disallows submit so that we don't need to worry about unwrapping exceptions on the
* tracing stage. See CASSANDRA-1123 for background. We allow submitting NO_OP tasks, to allow

16
src/java/org/apache/cassandra/config/Config.java
View File

@ -17,24 +17,21 @@
*/
package org.apache.cassandra.config;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.StringReader;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Supplier;
import com.google.common.base.Joiner;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -243,6 +240,7 @@ public class Config
public String memory_allocator;
private static boolean isClientMode = false;
private static Supplier<Config> overrideLoadConfig = null;
public Integer file_cache_size_in_mb;
@ -313,6 +311,16 @@ public class Config
isClientMode = clientMode;
}
public static Supplier<Config> getOverrideLoadConfig()
{
return overrideLoadConfig;
}
public static void setOverrideLoadConfig(Supplier<Config> loadConfig)
{
overrideLoadConfig = loadConfig;
}
public void configHintedHandoff() throws ConfigurationException
{
if (hinted_handoff_enabled != null && !hinted_handoff_enabled.isEmpty())

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

@ -128,6 +128,10 @@ public class DatabaseDescriptor
{
conf = new Config();
}
else if (Config.getOverrideLoadConfig() != null)
{
applyConfig(Config.getOverrideLoadConfig().get());
}
else
{
applyConfig(loadConfig());
@ -144,7 +148,7 @@ public class DatabaseDescriptor
{
String loaderClass = System.getProperty("cassandra.config.loader");
ConfigurationLoader loader = loaderClass == null
? new YamlConfigurationLoader()
? new YamlConfigurationLoader()
: FBUtilities.<ConfigurationLoader>construct(loaderClass, "configuration loading");
Config config = loader.loadConfig();

7
src/java/org/apache/cassandra/cql3/QueryOptions.java
View File

@ -83,7 +83,12 @@ public abstract class QueryOptions
public static QueryOptions create(ConsistencyLevel consistency, List<ByteBuffer> values, boolean skipMetadata, int pageSize, PagingState pagingState, ConsistencyLevel serialConsistency)
{
return new DefaultQueryOptions(consistency, values, skipMetadata, new SpecificOptions(pageSize, pagingState, serialConsistency, -1L), 0);
return create(consistency, values, skipMetadata, pageSize, pagingState, serialConsistency, 0);
}
public static QueryOptions create(ConsistencyLevel consistency, List<ByteBuffer> values, boolean skipMetadata, int pageSize, PagingState pagingState, ConsistencyLevel serialConsistency, int protocolVersion)
{
return new DefaultQueryOptions(consistency, values, skipMetadata, new SpecificOptions(pageSize, pagingState, serialConsistency, -1L), protocolVersion);
}
public static QueryOptions addColumnSpecifications(QueryOptions options, List<ColumnSpecification> columnSpecs)

8
src/java/org/apache/cassandra/cql3/QueryProcessor.java
View File

@ -156,7 +156,8 @@ public class QueryProcessor implements QueryHandler
}
}
private static QueryState internalQueryState()
@VisibleForTesting
public static QueryState internalQueryState()
{
return InternalStateInstance.INSTANCE.queryState;
}
@ -276,12 +277,13 @@ public class QueryProcessor implements QueryHandler
return null;
}
private static QueryOptions makeInternalOptions(ParsedStatement.Prepared prepared, Object[] values)
@VisibleForTesting
public static QueryOptions makeInternalOptions(ParsedStatement.Prepared prepared, Object[] values)
{
if (prepared.boundNames.size() != values.length)
throw new IllegalArgumentException(String.format("Invalid number of values. Expecting %d but got %d", prepared.boundNames.size(), values.length));
List<ByteBuffer> boundValues = new ArrayList<ByteBuffer>(values.length);
List<ByteBuffer> boundValues = new ArrayList<>(values.length);
for (int i = 0; i < values.length; i++)
{
Object value = values[i];

17
src/java/org/apache/cassandra/db/BlacklistedDirectories.java
View File

@ -21,15 +21,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.lang.management.ManagementFactory;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.MBeanWrapper;
public class BlacklistedDirectories implements BlacklistedDirectoriesMBean
{
@ -43,17 +40,7 @@ public class BlacklistedDirectories implements BlacklistedDirectoriesMBean
private BlacklistedDirectories()
{
// Register this instance with JMX
try
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
JVMStabilityInspector.inspectThrowable(e);
logger.error("error registering MBean {}", MBEAN_NAME, e);
//Allow the server to start even if the bean can't be registered
}
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME, MBeanWrapper.OnException.LOG);
}
public Set<File> getUnreadableDirectories()

36
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -19,6 +19,9 @@ package org.apache.cassandra.db;
import java.io.*;
import java.lang.management.ManagementFactory;
import java.io.File;
import java.io.IOException;
import java.io.PrintStream;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.*;
@ -189,12 +192,24 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public volatile long sampleLatencyNanos;
private final ScheduledFuture<?> latencyCalculator;
public static void shutdownFlushExecutor() throws InterruptedException
{
flushExecutor.shutdown();
flushExecutor.awaitTermination(60, TimeUnit.SECONDS);
}
public static void shutdownPostFlushExecutor() throws InterruptedException
{
postFlushExecutor.shutdown();
postFlushExecutor.awaitTermination(60, TimeUnit.SECONDS);
}
public static void shutdownReclaimExecutor() throws InterruptedException
{
reclaimExecutor.shutdown();
reclaimExecutor.awaitTermination(60, TimeUnit.SECONDS);
}
public void reload()
{
// metadata object has been mutated directly. make all the members jibe with new settings.
@ -423,16 +438,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
// register the mbean
String type = this.partitioner instanceof LocalPartitioner ? "IndexColumnFamilies" : "ColumnFamilies";
mbeanName = "org.apache.cassandra.db:type=" + type + ",keyspace=" + this.keyspace.getName() + ",columnfamily=" + name;
try
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
ObjectName nameObj = new ObjectName(mbeanName);
mbs.registerMBean(this, nameObj);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(this, mbeanName);
logger.trace("retryPolicy for {} is {}", name, this.metadata.getSpeculativeRetry());
latencyCalculator = ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(new Runnable()
{
@ -505,13 +511,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
data.removeUnreadableSSTables(directory);
}
void unregisterMBean() throws MalformedObjectNameException, InstanceNotFoundException, MBeanRegistrationException
void unregisterMBean()
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
ObjectName nameObj = new ObjectName(mbeanName);
if (mbs.isRegistered(nameObj))
mbs.unregisterMBean(nameObj);
if (MBeanWrapper.instance.isRegistered(mbeanName)) {
MBeanWrapper.instance.unregisterMBean(mbeanName);
}
// unregister metrics
metric.release();
}

16
src/java/org/apache/cassandra/db/HintedHandOffManager.java
View File

@ -19,15 +19,12 @@ package org.apache.cassandra.db;
import java.io.DataInputStream;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSortedSet;
@ -67,6 +64,9 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.WriteResponseHandler;
import org.apache.cassandra.utils.*;
import org.cliffc.high_scale_lib.NonBlockingHashSet;
import java.util.List;
import org.apache.cassandra.utils.MBeanWrapper;
/**
* The hint schema looks like this:
@ -164,15 +164,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
public void start()
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME);
logger.trace("Created HHOM instance, registered MBean.");
Runnable runnable = new Runnable()

3
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -52,7 +52,8 @@ public class Memtable implements Comparable<Memtable>
{
private static final Logger logger = LoggerFactory.getLogger(Memtable.class);
static final MemtablePool MEMORY_POOL = DatabaseDescriptor.getMemtableAllocatorPool();
@VisibleForTesting
public static final MemtablePool MEMORY_POOL = DatabaseDescriptor.getMemtableAllocatorPool();
private static final int ROW_OVERHEAD_HEAP_SIZE = estimateRowOverhead(Integer.parseInt(System.getProperty("cassandra.memtable_row_overhead_computation_step", "100000")));
private final MemtableAllocator allocator;

15
src/java/org/apache/cassandra/db/commitlog/CommitLog.java
View File

@ -18,13 +18,9 @@
package org.apache.cassandra.db.commitlog;
import java.io.*;
import java.lang.management.ManagementFactory;
import java.nio.ByteBuffer;
import java.util.*;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
@ -48,6 +44,7 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.CRC32Factory;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.MBeanWrapper;
import static org.apache.cassandra.db.commitlog.CommitLogSegment.*;
@ -77,15 +74,7 @@ public class CommitLog implements CommitLogMBean
{
CommitLog log = new CommitLog(DatabaseDescriptor.getCommitLogLocation(), CommitLogArchiver.construct());
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(log, new ObjectName("org.apache.cassandra.db:type=Commitlog"));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(log, "org.apache.cassandra.db:type=Commitlog");
return log.start();
}

20
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -19,11 +19,8 @@ package org.apache.cassandra.db.compaction;
import java.io.File;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.util.*;
import java.util.concurrent.*;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.management.openmbean.OpenDataException;
import javax.management.openmbean.TabularData;
@ -33,6 +30,7 @@ import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.collect.*;
import com.google.common.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -96,15 +94,8 @@ public class CompactionManager implements CompactionManagerMBean
static
{
instance = new CompactionManager();
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(instance, new ObjectName(MBEAN_OBJECT_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(instance, MBEAN_OBJECT_NAME);
}
private final CompactionExecutor executor = new CompactionExecutor();
@ -202,6 +193,7 @@ public class CompactionManager implements CompactionManagerMBean
// shutdown executors to prevent further submission
executor.shutdown();
validationExecutor.shutdown();
cacheCleanupExecutor.shutdown();
// interrupt compactions and validations
for (Holder compactionHolder : CompactionMetrics.getCompactions())
@ -212,7 +204,7 @@ public class CompactionManager implements CompactionManagerMBean
// wait for tasks to terminate
// compaction tasks are interrupted above, so it shuold be fairy quick
// until not interrupted tasks to complete.
for (ExecutorService exec : Arrays.asList(executor, validationExecutor))
for (ExecutorService exec : Arrays.asList(executor, validationExecutor, cacheCleanupExecutor))
{
try
{
@ -955,7 +947,7 @@ public class CompactionManager implements CompactionManagerMBean
public Bounded(final ColumnFamilyStore cfs, Collection<Range<Token>> ranges)
{
this.ranges = ranges;
cacheCleanupExecutor.submit(new Runnable()
instance.cacheCleanupExecutor.submit(new Runnable()
{
@Override
public void run()

14
src/java/org/apache/cassandra/gms/FailureDetector.java
View File

@ -18,15 +18,12 @@
package org.apache.cassandra.gms;
import java.io.*;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.*;
@ -37,6 +34,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MBeanWrapper;
/**
* This FailureDetector is an implementation of the paper titled
@ -81,15 +79,7 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean
public FailureDetector()
{
// Register this instance with JMX
try
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME);
}
private static long getInitialValue()

15
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -17,7 +17,6 @@
*/
package org.apache.cassandra.gms;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
@ -25,14 +24,12 @@ import java.util.Map.Entry;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -208,15 +205,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
FailureDetector.instance.registerFailureDetectionEventListener(this);
// Register this instance with JMX
try
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME);
}
public void setLastProcessedMessageAt(long timeInMillis)

26
src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
View File

@ -18,12 +18,14 @@
package org.apache.cassandra.io.sstable;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.util.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSet;
@ -34,12 +36,13 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableScheduledThreadPoolExecutor;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.WrappedRunnable;
@ -64,16 +67,7 @@ public class IndexSummaryManager implements IndexSummaryManagerMBean
static
{
instance = new IndexSummaryManager();
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(instance, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(instance, MBEAN_NAME);
}
private IndexSummaryManager()

24
src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.locator;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
@ -26,14 +25,13 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import com.codahale.metrics.ExponentiallyDecayingReservoir;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MBeanWrapper;
/**
@ -95,28 +93,12 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa
private void registerMBean()
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(this, new ObjectName(mbeanName));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(this, mbeanName);
}
public void unregisterMBean()
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.unregisterMBean(new ObjectName(mbeanName));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.unregisterMBean(mbeanName);
}
@Override

14
src/java/org/apache/cassandra/locator/EndpointSnitchInfo.java
View File

@ -18,28 +18,18 @@
package org.apache.cassandra.locator;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MBeanWrapper;
public class EndpointSnitchInfo implements EndpointSnitchInfoMBean
{
public static void create()
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(new EndpointSnitchInfo(), new ObjectName("org.apache.cassandra.db:type=EndpointSnitchInfo"));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(new EndpointSnitchInfo(), "org.apache.cassandra.db:type=EndpointSnitchInfo");
}
public String getDatacenter(String host) throws UnknownHostException

194
src/java/org/apache/cassandra/locator/InetAddressAndPort.java Normal file
View File

@ -0,0 +1,194 @@
/*
* 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.locator;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.UnknownHostException;
import com.google.common.base.Preconditions;
import com.google.common.net.HostAndPort;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.FastByteOperations;
/**
* A class to replace the usage of InetAddress to identify hosts in the cluster.
* Opting for a full replacement class so that in the future if we change the nature
* of the identifier the refactor will be easier in that we don't have to change the type
* just the methods.
*
* Because an IP might contain multiple C* instances the identification must be done
* using the IP + port. InetSocketAddress is undesirable for a couple of reasons. It's not comparable,
* it's toString() method doesn't correctly bracket IPv6, it doesn't handle optional default values,
* and a couple of other minor behaviors that are slightly less troublesome like handling the
* need to sometimes return a port and sometimes not.
*
*/
public final class InetAddressAndPort implements Comparable<InetAddressAndPort>, Serializable
{
private static final long serialVersionUID = 0;
//Store these here to avoid requiring DatabaseDescriptor to be loaded. DatabaseDescriptor will set
//these when it loads the config. A lot of unit tests won't end up loading DatabaseDescriptor.
//Tools that might use this class also might not load database descriptor. Those tools are expected
//to always override the defaults.
static volatile int defaultPort = 7000;
public final InetAddress address;
public final byte[] addressBytes;
public final int port;
private InetAddressAndPort(InetAddress address, byte[] addressBytes, int port)
{
Preconditions.checkNotNull(address);
Preconditions.checkNotNull(addressBytes);
validatePortRange(port);
this.address = address;
this.port = port;
this.addressBytes = addressBytes;
}
private static void validatePortRange(int port)
{
if (port < 0 | port > 65535)
{
throw new IllegalArgumentException("Port " + port + " is not a valid port number in the range 0-65535");
}
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
InetAddressAndPort that = (InetAddressAndPort) o;
if (port != that.port) return false;
return address.equals(that.address);
}
@Override
public int hashCode()
{
int result = address.hashCode();
result = 31 * result + port;
return result;
}
@Override
public int compareTo(InetAddressAndPort o)
{
int retval = FastByteOperations.compareUnsigned(addressBytes, 0, addressBytes.length, o.addressBytes, 0, o.addressBytes.length);
if (retval != 0)
{
return retval;
}
return Integer.compare(port, o.port);
}
public String getHostAddress(boolean withPort)
{
if (withPort)
{
return toString();
}
else
{
return address.getHostAddress();
}
}
@Override
public String toString()
{
return toString(true);
}
public String toString(boolean withPort)
{
if (withPort)
{
return HostAndPort.fromParts(address.getHostAddress(), port).toString();
}
else
{
return address.toString();
}
}
public static InetAddressAndPort getByName(String name) throws UnknownHostException
{
return getByNameOverrideDefaults(name, null);
}
/**
*
* @param name Hostname + optional ports string
* @param port Port to connect on, overridden by values in hostname string, defaults to DatabaseDescriptor default if not specified anywhere.
* @return
* @throws UnknownHostException
*/
public static InetAddressAndPort getByNameOverrideDefaults(String name, Integer port) throws UnknownHostException
{
HostAndPort hap = HostAndPort.fromString(name);
if (hap.hasPort())
{
port = hap.getPort();
}
return getByAddressOverrideDefaults(InetAddress.getByName(hap.getHostText()), port);
}
public static InetAddressAndPort getByAddress(byte[] address) throws UnknownHostException
{
return getByAddressOverrideDefaults(InetAddress.getByAddress(address), address, null);
}
public static InetAddressAndPort getByAddress(InetAddress address)
{
return getByAddressOverrideDefaults(address, null);
}
public static InetAddressAndPort getByAddressOverrideDefaults(InetAddress address, Integer port)
{
if (port == null)
{
port = defaultPort;
}
return new InetAddressAndPort(address, address.getAddress(), port);
}
public static InetAddressAndPort getByAddressOverrideDefaults(InetAddress address, byte[] addressBytes, Integer port)
{
if (port == null)
{
port = defaultPort;
}
return new InetAddressAndPort(address, addressBytes, port);
}
public static void initializeDefaultPort(int port)
{
defaultPort = port;
}
}

11
src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java
View File

@ -17,13 +17,15 @@
*/
package org.apache.cassandra.metrics;
import java.lang.management.ManagementFactory;
import java.lang.reflect.Method;
import java.util.Locale;
import java.util.concurrent.TimeUnit;
import com.codahale.metrics.*;
import javax.management.*;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.apache.cassandra.utils.MBeanWrapper;
/**
* Makes integrating 3.0 metrics API with 2.0.
@ -35,7 +37,7 @@ public class CassandraMetricsRegistry extends MetricRegistry
{
public static final CassandraMetricsRegistry Metrics = new CassandraMetricsRegistry();
private final MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
private final MBeanWrapper mBeanServer = MBeanWrapper.instance;
private CassandraMetricsRegistry()
{
@ -128,7 +130,8 @@ public class CassandraMetricsRegistry extends MetricRegistry
try
{
mBeanServer.registerMBean(mbean, name);
} catch (Exception ignored) {}
}
catch (Exception ignored) {}
}
public interface MetricMBean

13
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.net;
import java.io.*;
import java.lang.management.ManagementFactory;
import java.net.*;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
@ -29,8 +28,6 @@ import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.net.ssl.SSLHandshakeException;
import com.google.common.annotations.VisibleForTesting;
@ -409,15 +406,7 @@ public final class MessagingService implements MessagingServiceMBean
if (!testOnly)
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME);
}
}

15
src/java/org/apache/cassandra/service/CacheService.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.service;
import java.io.DataInputStream;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
@ -28,8 +27,6 @@ import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.util.concurrent.Futures;
@ -52,6 +49,7 @@ import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.Pair;
public class CacheService implements CacheServiceMBean
@ -87,16 +85,7 @@ public class CacheService implements CacheServiceMBean
private CacheService()
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME);
keyCache = initKeyCache();
rowCache = initRowCache();

11
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -536,16 +536,7 @@ public class CassandraDaemon
throw e.getCause();
}
try
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
mbs.registerMBean(new StandardMBean(new NativeAccess(), NativeAccessMBean.class), new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
logger.error("error registering MBean {}", MBEAN_NAME, e);
//Allow the server to start even if the bean can't be registered
}
MBeanWrapper.instance.registerMBean(new StandardMBean(new NativeAccess(), NativeAccessMBean.class), MBEAN_NAME, MBeanWrapper.OnException.LOG);
if (FBUtilities.isWindows())
{

9
src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java
View File

@ -30,6 +30,8 @@ import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.annotations.VisibleForTesting;
public class PendingRangeCalculatorService
{
public static final PendingRangeCalculatorService instance = new PendingRangeCalculatorService();
@ -108,4 +110,11 @@ public class PendingRangeCalculatorService
{
StorageService.instance.getTokenMetadata().calculatePendingRanges(strategy, keyspaceName);
}
@VisibleForTesting
public void shutdownExecutor() throws InterruptedException
{
executor.shutdown();
executor.awaitTermination(60, TimeUnit.SECONDS);
}
}

14
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -18,14 +18,12 @@
package org.apache.cassandra.service;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.base.Predicate;
import com.google.common.cache.CacheLoader;
@ -98,15 +96,7 @@ public class StorageProxy implements StorageProxyMBean
static
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(instance, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
MBeanWrapper.instance.registerMBean(instance, MBEAN_NAME);
standardWritePerformer = new WritePerformer()
{

16
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -179,7 +179,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
private static final BackgroundActivityMonitor bgMonitor = new BackgroundActivityMonitor();
private final ObjectName jmxObjectName;
private final String jmxObjectName;
private Collection<Token> bootstrapTokens = null;
@ -230,17 +230,9 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
// use dedicated executor for sending JMX notifications
super(Executors.newSingleThreadExecutor());
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
jmxObjectName = new ObjectName("org.apache.cassandra.db:type=StorageService");
mbs.registerMBean(this, jmxObjectName);
mbs.registerMBean(StreamManager.instance, new ObjectName(StreamManager.OBJECT_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
jmxObjectName = "org.apache.cassandra.db:type=StorageService";
MBeanWrapper.instance.registerMBean(this, jmxObjectName);
MBeanWrapper.instance.registerMBean(StreamManager.instance, StreamManager.OBJECT_NAME);
legacyProgressSupport = new LegacyJMXProgressSupport(this, jmxObjectName);

31
src/java/org/apache/cassandra/utils/ByteBufferUtil.java
View File

@ -412,6 +412,37 @@ public class ByteBufferUtil
return bytes.getDouble(bytes.position());
}
public static ByteBuffer objectToBytes(Object obj)
{
if (obj instanceof Integer)
return ByteBufferUtil.bytes((int) obj);
else if (obj instanceof Byte)
return ByteBufferUtil.bytes((byte) obj);
else if (obj instanceof Short)
return ByteBufferUtil.bytes((short) obj);
else if (obj instanceof Long)
return ByteBufferUtil.bytes((long) obj);
else if (obj instanceof Float)
return ByteBufferUtil.bytes((float) obj);
else if (obj instanceof Double)
return ByteBufferUtil.bytes((double) obj);
else if (obj instanceof UUID)
return ByteBufferUtil.bytes((UUID) obj);
else if (obj instanceof InetAddress)
return ByteBufferUtil.bytes((InetAddress) obj);
else if (obj instanceof String)
return ByteBufferUtil.bytes((String) obj);
else
throw new IllegalArgumentException(String.format("Cannot convert value %s of type %s",
obj,
obj.getClass()));
}
public static ByteBuffer bytes(byte b)
{
return ByteBuffer.allocate(1).put(0, b);
}
public static ByteBuffer bytes(short s)
{
return ByteBuffer.allocate(2).putShort(0, s);

209
src/java/org/apache/cassandra/utils/MBeanWrapper.java Normal file
View File

@ -0,0 +1,209 @@
/*
* 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.utils;
import java.lang.management.ManagementFactory;
import java.util.function.Consumer;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Helper class to avoid catching and rethrowing checked exceptions on MBean and
* allow turning of MBean registration for test purposes.
*/
public interface MBeanWrapper
{
static final Logger logger = LoggerFactory.getLogger(MBeanWrapper.class);
static final MBeanWrapper instance = Boolean.getBoolean("org.apache.cassandra.disable_mbean_registration") ?
new NoOpMBeanWrapper() :
new PlatformMBeanWrapper();
// Passing true for graceful will log exceptions instead of rethrowing them
public void registerMBean(Object obj, ObjectName mbeanName, OnException onException);
public void registerMBean(Object obj, ObjectName mbeanName);
public void registerMBean(Object obj, String mbeanName, OnException onException);
public void registerMBean(Object obj, String mbeanName);
public boolean isRegistered(ObjectName mbeanName, OnException onException);
public boolean isRegistered(ObjectName mbeanName);
public boolean isRegistered(String mbeanName, OnException onException);
public boolean isRegistered(String mbeanName);
public void unregisterMBean(ObjectName mbeanName, OnException onException);
public void unregisterMBean(ObjectName mbeanName);
public void unregisterMBean(String mbeanName, OnException onException);
public void unregisterMBean(String mbeanName);
static class NoOpMBeanWrapper implements MBeanWrapper
{
public void registerMBean(Object obj, ObjectName mbeanName, OnException onException) {}
public void registerMBean(Object obj, ObjectName mbeanName) {}
public void registerMBean(Object obj, String mbeanName, OnException onException) {}
public void registerMBean(Object obj, String mbeanName) {}
public boolean isRegistered(ObjectName mbeanName, OnException onException) { return false; }
public boolean isRegistered(ObjectName mbeanName) { return false; }
public boolean isRegistered(String mbeanName, OnException onException) { return false; }
public boolean isRegistered(String mbeanName) { return false; }
public void unregisterMBean(ObjectName mbeanName, OnException onException) {}
public void unregisterMBean(ObjectName mbeanName) {}
public void unregisterMBean(String mbeanName, OnException onException) {}
public void unregisterMBean(String mbeanName) {}
}
static class PlatformMBeanWrapper implements MBeanWrapper
{
private final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
public void registerMBean(Object obj, ObjectName mbeanName, OnException onException)
{
try
{
mbs.registerMBean(obj, mbeanName);
}
catch (Exception e)
{
onException.handler.accept(e);
}
}
public void registerMBean(Object obj, ObjectName mbeanName)
{
registerMBean(obj, mbeanName, OnException.THROW);
}
public void registerMBean(Object obj, String mbeanName, OnException onException)
{
try
{
mbs.registerMBean(obj, new ObjectName(mbeanName));
}
catch (Exception e)
{
onException.handler.accept(e);
}
}
public void registerMBean(Object obj, String mbeanName)
{
registerMBean(obj, mbeanName, OnException.THROW);
}
public boolean isRegistered(ObjectName mbeanName, OnException onException)
{
try
{
return mbs.isRegistered(mbeanName);
}
catch (Exception e)
{
onException.handler.accept(e);
}
return false;
}
public boolean isRegistered(ObjectName mbeanName)
{
return isRegistered(mbeanName, OnException.THROW);
}
public boolean isRegistered(String mbeanName, OnException onException)
{
try
{
return mbs.isRegistered(new ObjectName(mbeanName));
}
catch (Exception e)
{
onException.handler.accept(e);
}
return false;
}
public boolean isRegistered(String mbeanName)
{
return isRegistered(mbeanName, OnException.THROW);
}
public void unregisterMBean(ObjectName mbeanName, OnException onException)
{
try
{
mbs.unregisterMBean(mbeanName);
}
catch (Exception e)
{
onException.handler.accept(e);
}
}
public void unregisterMBean(ObjectName mbeanName)
{
unregisterMBean(mbeanName, OnException.THROW);
}
public void unregisterMBean(String mbeanName, OnException onException)
{
try
{
mbs.unregisterMBean(new ObjectName(mbeanName));
}
catch (Exception e)
{
onException.handler.accept(e);
}
}
public void unregisterMBean(String mbeanName)
{
unregisterMBean(mbeanName, OnException.THROW);
}
}
public enum OnException
{
THROW(new Consumer<Exception>()
{
public void accept(Exception e)
{
throw new RuntimeException(e);
}
}),
LOG(new Consumer<Exception>()
{
public void accept(Exception e)
{
logger.error("Error in MBean wrapper: ", e);
}
}),
IGNORE(new Consumer<Exception>()
{
public void accept(Exception e)
{
}
});
private Consumer<Exception> handler;
OnException(Consumer<Exception> handler)
{
this.handler = handler;
}
}
}

4
src/java/org/apache/cassandra/utils/Mx4jTool.java
View File

@ -17,8 +17,6 @@
*/
package org.apache.cassandra.utils;
import java.lang.management.ManagementFactory;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.slf4j.Logger;
@ -43,7 +41,7 @@ public class Mx4jTool
try
{
logger.trace("Will try to load mx4j now, if it's in the classpath");
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
MBeanWrapper mbs = MBeanWrapper.instance;
ObjectName processorName = new ObjectName("Server:name=XSLTProcessor");
Class<?> httpAdaptorClass = Class.forName("mx4j.tools.adaptor.http.HttpAdaptor");

47
src/java/org/apache/cassandra/utils/concurrent/Ref.java
View File

@ -29,9 +29,12 @@ import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.concurrent.InfiniteLoopExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.InfiniteLoopExecutor.InterruptibleRunnable;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import static org.apache.cassandra.utils.Throwables.maybeFail;
@ -323,32 +326,26 @@ public final class Ref<T> implements RefCounted<T>
private static final Set<GlobalState> globallyExtant = Collections.newSetFromMap(new ConcurrentHashMap<GlobalState, Boolean>());
static final ReferenceQueue<Object> referenceQueue = new ReferenceQueue<>();
private static final ExecutorService EXEC = Executors.newFixedThreadPool(1, new NamedThreadFactory("Reference-Reaper"));
static
private static final InfiniteLoopExecutor EXEC = new InfiniteLoopExecutor("Reference-Reaper", new InterruptibleRunnable()
{
EXEC.execute(new Runnable()
public void run() throws InterruptedException
{
public void run()
{
try
{
while (true)
{
Object obj = referenceQueue.remove();
if (obj instanceof Ref.State)
{
((Ref.State) obj).release(true);
}
}
}
catch (InterruptedException e)
{
}
finally
{
EXEC.execute(this);
}
}
});
reapOneReference();
}
}).start();
private static void reapOneReference() throws InterruptedException
{
Object obj = referenceQueue.remove(100);
if (obj instanceof Ref.State)
{
((Ref.State) obj).release(true);
}
}
@VisibleForTesting
public static void shutdownReferenceReaper() throws InterruptedException
{
EXEC.shutdown();
EXEC.awaitTermination(60, TimeUnit.SECONDS);
}
}

85
src/java/org/apache/cassandra/utils/memory/MemtableCleanerThread.java
View File

@ -18,57 +18,76 @@
*/
package org.apache.cassandra.utils.memory;
import org.apache.cassandra.concurrent.InfiniteLoopExecutor;
import org.apache.cassandra.utils.concurrent.WaitQueue;
/**
* A thread that reclaims memory from a MemtablePool on demand. The actual reclaiming work is delegated to the
* cleaner Runnable, e.g., FlushLargestColumnFamily
*/
class MemtableCleanerThread<P extends MemtablePool> extends Thread
public class MemtableCleanerThread<P extends MemtablePool> extends InfiniteLoopExecutor
{
/** The pool we're cleaning */
final P pool;
/** should ensure that at least some memory has been marked reclaiming after completion */
final Runnable cleaner;
/** signalled whenever needsCleaning() may return true */
final WaitQueue wait = new WaitQueue();
MemtableCleanerThread(P pool, Runnable cleaner)
private static class Clean<P extends MemtablePool> implements InterruptibleRunnable
{
super(pool.getClass().getSimpleName() + "Cleaner");
this.pool = pool;
this.cleaner = cleaner;
setDaemon(true);
}
/** The pool we're cleaning */
final P pool;
boolean needsCleaning()
{
return pool.onHeap.needsCleaning() || pool.offHeap.needsCleaning();
}
/** should ensure that at least some memory has been marked reclaiming after completion */
final Runnable cleaner;
// should ONLY be called when we really think it already needs cleaning
void trigger()
{
wait.signal();
}
/** signalled whenever needsCleaning() may return true */
final WaitQueue wait = new WaitQueue();
@Override
public void run()
{
while (true)
private Clean(P pool, Runnable cleaner)
{
while (!needsCleaning())
this.pool = pool;
this.cleaner = cleaner;
}
boolean needsCleaning()
{
return pool.onHeap.needsCleaning() || pool.offHeap.needsCleaning();
}
@Override
public void run() throws InterruptedException
{
if (needsCleaning())
{
cleaner.run();
}
else
{
final WaitQueue.Signal signal = wait.register();
if (!needsCleaning())
signal.awaitUninterruptibly();
signal.await();
else
signal.cancel();
}
cleaner.run();
}
}
private final Runnable trigger;
private MemtableCleanerThread(final Clean<P> clean)
{
super(clean.pool.getClass().getSimpleName() + "Cleaner", clean);
this.trigger = new Runnable()
{
public void run()
{
clean.wait.signal();
}
};
}
MemtableCleanerThread(P pool, Runnable cleaner)
{
this(new Clean<>(pool, cleaner));
}
// should ONLY be called when we really think it already needs cleaning
public void trigger()
{
trigger.run();
}
}

7
src/java/org/apache/cassandra/utils/memory/MemtablePool.java
View File

@ -18,6 +18,7 @@
*/
package org.apache.cassandra.utils.memory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import org.apache.cassandra.utils.concurrent.WaitQueue;
@ -57,6 +58,12 @@ public abstract class MemtablePool
}
public abstract boolean needToCopyOnHeap();
public void shutdown() throws InterruptedException
{
cleaner.shutdown();
cleaner.awaitTermination(60, TimeUnit.SECONDS);
}
public abstract MemtableAllocator newAllocator();
/**

4
src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java
View File

@ -40,12 +40,12 @@ public class LegacyJMXProgressSupport implements ProgressListener
protected static final Pattern SESSION_SUCCESS_MATCHER = Pattern.compile("Repair session .* for range .* finished");
private final AtomicLong notificationSerialNumber = new AtomicLong();
private final ObjectName jmxObjectName;
private final String jmxObjectName;
private final NotificationBroadcasterSupport broadcaster;
public LegacyJMXProgressSupport(NotificationBroadcasterSupport broadcaster,
ObjectName jmxObjectName)
String jmxObjectName)
{
this.broadcaster = broadcaster;
this.jmxObjectName = jmxObjectName;

79
test/conf/logback-dtest.xml Normal file
View File

@ -0,0 +1,79 @@
<!--
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.
-->
<configuration debug="false" scan="true" scanPeriod="60 seconds">
<define name="instance_id" class="org.apache.cassandra.distributed.InstanceIDDefiner" />
<!-- Shutdown hook ensures that async appender flushes -->
<shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/>
<!-- Status listener is used to wrap stdout/stderr and tee to log file -->
<statusListener class="org.apache.cassandra.LogbackStatusListener" />
<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>./build/test/logs/${cassandra.testtag}/TEST-${suitename}.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
<fileNamePattern>./build/test/logs/${cassandra.testtag}/TEST-${suitename}.log.%i.gz</fileNamePattern>
<minIndex>1</minIndex>
<maxIndex>20</maxIndex>
</rollingPolicy>
<triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
<maxFileSize>20MB</maxFileSize>
</triggeringPolicy>
<encoder>
<pattern>%-5level [%thread] ${instance_id} %date{ISO8601} %msg%n</pattern>
<immediateFlush>false</immediateFlush>
</encoder>
</appender>
<appender name="STDOUT" target="System.out" class="org.apache.cassandra.ConsoleAppender">
<encoder>
<pattern>%-5level [%thread] ${instance_id} %date{ISO8601} %F:%L - %msg%n</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>DEBUG</level>
</filter>
</appender>
<appender name="TEE" class="org.apache.cassandra.TeeingAppender">
<appender-ref ref="FILE"/>
<appender-ref ref="STDOUT"/>
</appender>
<logger name="org.apache.hadoop" level="WARN"/>
<logger name="org.apache.cassandra.db.monitoring" level="DEBUG"/>
<!-- Do not change the name of this appender. LogbackStatusListener uses the thread name
tied to the appender name to know when to write to real stdout/stderr vs forwarding to logback -->
<appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">
<discardingThreshold>0</discardingThreshold>
<maxFlushTime>0</maxFlushTime>
<queueSize>1024</queueSize>
<appender-ref ref="TEE"/>
<includeCallerData>true</includeCallerData>
</appender>
<root level="DEBUG">
<appender-ref ref="ASYNC" />
</root>
</configuration>

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

@ -0,0 +1,79 @@
/*
* 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));
}
prepared.validate(QueryState.forInternalCalls().getClientState());
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);
}
}

92
test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java Normal file
View File

@ -0,0 +1,92 @@
/*
* 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 org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.db.ConsistencyLevel;
public class DistributedReadWritePathTest extends DistributedTestBase
{
@Test
public void coordinatorRead() throws Throwable
{
try (TestCluster cluster = createCluster(3))
{
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)");
assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?",
ConsistencyLevel.ALL,
1),
row(1, 1, 1),
row(1, 2, 2),
row(1, 3, 3));
}
}
@Test
public void coordinatorWrite() throws Throwable
{
try (TestCluster cluster = createCluster(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)",
ConsistencyLevel.QUORUM);
for (int i = 0; i < 3; i++)
{
assertRows(cluster.get(1).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"),
row(1, 1, 1));
}
assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
ConsistencyLevel.QUORUM),
row(1, 1, 1));
}
}
@Test
public void readRepairTest() throws Throwable
{
try (TestCluster cluster = createCluster(3))
{
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, 1, 1)");
assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"));
assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
ConsistencyLevel.QUORUM),
row(1, 1, 1));
// Verify that data got repaired to the third node
assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"),
row(1, 1, 1));
}
}
}

86
test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java Normal file
View File

@ -0,0 +1,86 @@
/*
* 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.util.Arrays;
import org.junit.Assert;
import org.junit.BeforeClass;
public class DistributedTestBase
{
static String KEYSPACE = "distributed_test_keyspace";
@BeforeClass
public static void setup()
{
System.setProperty("org.apache.cassandra.disable_mbean_registration", "true");
}
TestCluster createCluster(int nodeCount) throws Throwable
{
TestCluster cluster = TestCluster.create(nodeCount);
cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + nodeCount + "};");
return cluster;
}
public static void assertRows(Object[][] actual, Object[]... expected)
{
Assert.assertEquals(rowsNotEqualErrorMessage(actual, expected),
expected.length, actual.length);
for (int i = 0; i < expected.length; i++)
{
Object[] expectedRow = expected[i];
Object[] actualRow = actual[i];
Assert.assertTrue(rowsNotEqualErrorMessage(actual, expected),
Arrays.equals(expectedRow, actualRow));
}
}
public static String rowsNotEqualErrorMessage(Object[][] actual, Object[][] expected)
{
return String.format("Expected: %s\nActual:%s\n",
rowsToString(expected),
rowsToString(actual));
}
public static String rowsToString(Object[][] rows)
{
StringBuilder builder = new StringBuilder();
builder.append("[");
boolean isFirst = true;
for (Object[] row : rows)
{
if (isFirst)
isFirst = false;
else
builder.append(",");
builder.append(Arrays.toString(row));
}
builder.append("]");
return builder.toString();
}
public static Object[] row(Object... expected)
{
return expected;
}
}

371
test/distributed/org/apache/cassandra/distributed/Instance.java Normal file
View File

@ -0,0 +1,371 @@
/*
* 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.DataInputStream;
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;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.function.BiConsumer;
import java.util.function.Function;
import org.apache.cassandra.concurrent.ScheduledExecutors;
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;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.statements.ParsedStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Memtable;
import org.apache.cassandra.db.SystemKeyspace;
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.gms.ApplicationState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.VersionedValue;
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.service.ClientState;
import org.apache.cassandra.service.PendingRangeCalculatorService;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.concurrent.Ref;
public class Instance extends InvokableInstance
{
public final InstanceConfig config;
public Instance(InstanceConfig config, ClassLoader classLoader)
{
super(classLoader);
this.config = config;
}
public InetAddressAndPort getBroadcastAddress() { return callOnInstance(LegacyAdapter::getBroadcastAddressAndPort); }
public Object[][] executeInternal(String query, Object... args)
{
return callOnInstance(() ->
{
ParsedStatement.Prepared prepared = QueryProcessor.prepareInternal(query);
ResultMessage result = prepared.statement.executeInternal(QueryProcessor.internalQueryState(),
QueryProcessor.makeInternalOptions(prepared, args));
if (result instanceof ResultMessage.Rows)
return RowUtil.toObjects((ResultMessage.Rows)result);
else
return null;
});
}
public UUID getSchemaVersion()
{
// we do not use method reference syntax here, because we need to invoke on the node-local schema instance
//noinspection Convert2MethodRef
return callOnInstance(() -> Schema.instance.getVersion());
}
public void schemaChange(String query)
{
runOnInstance(() ->
{
try
{
ClientState state = ClientState.forInternalCalls();
state.setKeyspace(SystemKeyspace.NAME);
QueryState queryState = new QueryState(state);
CQLStatement statement = QueryProcessor.parseStatement(query, queryState).statement;
statement.validate(state);
QueryOptions options = QueryOptions.forInternalCalls(Collections.emptyList());
statement.executeInternal(queryState, options);
}
catch (Exception e)
{
throw new RuntimeException("Error setting schema for test (query was: " + query + ")", e);
}
});
}
private void registerMockMessaging(TestCluster cluster)
{
BiConsumer<InetAddressAndPort, Message> deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message);
BiConsumer<InetAddressAndPort, Message> 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");
});
acceptsOnInstance((BiConsumer<InetAddressAndPort, Message> deliver) ->
MessagingService.instance().addMessageSink(new MessageDeliverySink(deliver, addressAndPortMap::get))
).accept(deliverToInstanceIfNotFiltered);
}
private static class MessageDeliverySink implements IMessageSink
{
private final BiConsumer<InetAddressAndPort, Message> deliver;
private final Function<InetAddress, InetAddressAndPort> lookupAddressAndPort;
MessageDeliverySink(BiConsumer<InetAddressAndPort, Message> deliver, Function<InetAddress, InetAddressAndPort> lookupAddressAndPort)
{
this.deliver = deliver;
this.lookupAddressAndPort = lookupAddressAndPort;
}
public boolean allowOutgoingMessage(MessageOut messageOut, int id, InetAddress to)
{
try (DataOutputBuffer out = new DataOutputBuffer(1024))
{
InetAddressAndPort from = LegacyAdapter.getBroadcastAddressAndPort();
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));
}
catch (IOException e)
{
throw new RuntimeException(e);
}
return false;
}
public boolean allowIncomingMessage(MessageIn message, int id)
{
// we can filter to our heart's content on the outgoing message; no need to worry about incoming
return true;
}
}
private void receiveMessage(Message message)
{
acceptsOnInstance((Message m) ->
{
try (DataInputStream in = new DataInputStream(new ByteArrayInputStream(m.bytes)))
{
MessageIn<?> messageIn = MessageIn.read(in, m.version, m.id);
Runnable deliver = new MessageDeliveryTask(messageIn, m.id, System.currentTimeMillis(), false);
deliver.run();
}
catch (Throwable t)
{
throw new RuntimeException("Exception occurred on the node " + LegacyAdapter.getBroadcastAddressAndPort(), t);
}
}).accept(message);
}
void launch(TestCluster cluster)
{
try
{
mkdirs();
int id = config.num;
runOnInstance(() -> InstanceIDDefiner.instanceId = id); // for logging
startup();
initializeRing(cluster);
registerMockMessaging(cluster);
}
catch (Throwable t)
{
if (t instanceof RuntimeException)
throw (RuntimeException) t;
throw new RuntimeException(t);
}
}
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(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)
{
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.concurrent_writes = overrides.concurrent_writes;
config.concurrent_counter_writes = overrides.concurrent_counter_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;
return config;
}
private void initializeRing(TestCluster cluster)
{
// This should be done outside instance in order to avoid serializing config
String partitionerName = config.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);
}
runOnInstance(() ->
{
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));
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
{
throw new RuntimeException(e);
}
});
}
void shutdown()
{
runOnInstance(() -> {
Throwable error = null;
error = runAndMergeThrowable(error,
Gossiper.instance::stop,
CompactionManager.instance::forceShutdown,
CommitLog.instance::shutdownBlocking,
ColumnFamilyStore::shutdownFlushExecutor,
ColumnFamilyStore::shutdownPostFlushExecutor,
ColumnFamilyStore::shutdownReclaimExecutor,
PendingRangeCalculatorService.instance::shutdownExecutor,
Ref::shutdownReferenceReaper,
Memtable.MEMORY_POOL::shutdown,
StageManager::shutdownAndWait,
SharedExecutorPool::shutdownSharedPool,
MessagingService.instance()::shutdown,
ScheduledExecutors::shutdownAndWait);
Throwables.maybeFail(error);
});
}
private static Throwable runAndMergeThrowable(Throwable existing, ThrowingRunnable ... runnables)
{
for (ThrowingRunnable runnable : runnables)
{
try
{
runnable.run();
}
catch (Throwable t)
{
existing = Throwables.merge(existing, t);
}
}
return existing;
}
public static interface ThrowingRunnable
{
public void run() throws Throwable;
}
}

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

@ -0,0 +1,101 @@
/*
* 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);
}
}

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

@ -0,0 +1,87 @@
/*
* 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);
}
}

38
test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java Normal file
View File

@ -0,0 +1,38 @@
/*
* 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 ch.qos.logback.core.PropertyDefinerBase;
/**
* Used by logback to find/define property value, see logback-dtest.xml
*/
public class InstanceIDDefiner extends PropertyDefinerBase
{
// Instantiated per classloader, set by Instance
public static int instanceId = -1;
public String getPropertyValue()
{
if (instanceId == -1)
return "<main>";
else
return "INSTANCE" + instanceId;
}
}

133
test/distributed/org/apache/cassandra/distributed/InvokableInstance.java Normal file
View File

@ -0,0 +1,133 @@
/*
* 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.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.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
public abstract class InvokableInstance
{
private final ClassLoader classLoader;
private final Method deserializeOnInstance;
public InvokableInstance(ClassLoader classLoader)
{
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 SerializableCallable<T> extends Callable<T>, Serializable { public T call(); }
public <T> SerializableCallable<T> callsOnInstance(SerializableCallable<T> call) { return (SerializableCallable<T>) transferOneObject(call); }
public <T> T callOnInstance(SerializableCallable<T> call) { return callsOnInstance(call).call(); }
public interface SerializableRunnable extends Runnable, Serializable {}
public SerializableRunnable runsOnInstance(SerializableRunnable run) { return (SerializableRunnable) transferOneObject(run); }
public void runOnInstance(SerializableRunnable run) { runsOnInstance(run).run(); }
public interface SerializableConsumer<T> extends Consumer<T>, Serializable {}
public <T> SerializableConsumer<T> acceptsOnInstance(SerializableConsumer<T> consumer) { return (SerializableConsumer<T>) transferOneObject(consumer); }
public interface SerializableBiConsumer<T1, T2> extends BiConsumer<T1, T2>, Serializable {}
public <T1, T2> SerializableBiConsumer<T1, T2> acceptsOnInstance(SerializableBiConsumer<T1, T2> consumer) { return (SerializableBiConsumer<T1, T2>) transferOneObject(consumer); }
public interface SerializableFunction<I, O> extends Function<I, O>, Serializable {}
public <I, O> SerializableFunction<I, O> appliesOnInstance(SerializableFunction<I, O> f) { return (SerializableFunction<I, O>) transferOneObject(f); }
public interface SerializableBiFunction<I1, I2, O> extends BiFunction<I1, I2, O>, Serializable {}
public <I1, I2, O> SerializableBiFunction<I1, I2, O> appliesOnInstance(SerializableBiFunction<I1, I2, O> f) { return (SerializableBiFunction<I1, I2, O>) transferOneObject(f); }
public interface SerializableTriFunction<I1, I2, I3, O> extends Serializable
{
O apply(I1 i1, I2 i2, I3 i3);
}
public <I1, I2, I3, O> SerializableTriFunction<I1, I2, I3, O> appliesOnInstance(SerializableTriFunction<I1, I2, I3, O> f) { return (SerializableTriFunction<I1, I2, I3, O>) transferOneObject(f); }
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);
}
}
}

42
test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java Normal file
View File

@ -0,0 +1,42 @@
/*
* 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 org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.utils.FBUtilities;
import java.net.InetAddress;
public class LegacyAdapter
{
private static final InetAddressAndPort broadcastAddressAndPort;
static
{
InetAddress address = FBUtilities.getBroadcastAddress();
int port = DatabaseDescriptor.getStoragePort();
broadcastAddressAndPort = InetAddressAndPort.getByAddressOverrideDefaults(address, port);
}
public static InetAddressAndPort getBroadcastAddressAndPort()
{
return broadcastAddressAndPort;
}
}

41
test/distributed/org/apache/cassandra/distributed/Message.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;
import org.apache.cassandra.locator.InetAddressAndPort;
// a container for simplifying the method signature for per-instance message handling/delivery
public class Message
{
public final int verb;
public final byte[] bytes;
public final int id;
public final int version;
public final InetAddressAndPort from;
public Message(int verb, byte[] bytes, int id, int version, InetAddressAndPort from)
{
this.verb = verb;
this.bytes = bytes;
this.id = id;
this.version = version;
this.from = from;
}
}

175
test/distributed/org/apache/cassandra/distributed/MessageFilters.java Normal file
View File

@ -0,0 +1,175 @@
/*
* 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.util.Arrays;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.function.BiConsumer;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessagingService;
public class MessageFilters
{
private final TestCluster cluster;
private final Set<Filter> filters = new CopyOnWriteArraySet<>();
public MessageFilters(TestCluster cluster)
{
this.cluster = cluster;
}
BiConsumer<InetAddressAndPort, Message> filter(BiConsumer<InetAddressAndPort, Message> applyIfNotFiltered)
{
return (toAddress, message) ->
{
int from = cluster.get(message.from).config.num;
int to = cluster.get(toAddress).config.num;
int verb = message.verb;
for (Filter filter : filters)
{
if (filter.matches(from, to, verb))
return;
}
applyIfNotFiltered.accept(toAddress, message);
};
}
public class Filter
{
final int[] from;
final int[] to;
final int[] verbs;
Filter(int[] from, int[] to, int[] verbs)
{
if (from != null)
{
from = from.clone();
Arrays.sort(from);
}
if (to != null)
{
to = to.clone();
Arrays.sort(to);
}
if (verbs != null)
{
verbs = verbs.clone();
Arrays.sort(verbs);
}
this.from = from;
this.to = to;
this.verbs = verbs;
}
public int hashCode()
{
return (from == null ? 0 : Arrays.hashCode(from))
+ (to == null ? 0 : Arrays.hashCode(to))
+ (verbs == null ? 0 : Arrays.hashCode(verbs));
}
public boolean equals(Object that)
{
return that instanceof Filter && equals((Filter) that);
}
public boolean equals(Filter that)
{
return Arrays.equals(from, that.from)
&& Arrays.equals(to, that.to)
&& Arrays.equals(verbs, that.verbs);
}
public boolean matches(int from, int to, int verb)
{
return (this.from == null || Arrays.binarySearch(this.from, from) >= 0)
&& (this.to == null || Arrays.binarySearch(this.to, to) >= 0)
&& (this.verbs == null || Arrays.binarySearch(this.verbs, verb) >= 0);
}
public Filter restore()
{
filters.remove(this);
return this;
}
public Filter drop()
{
filters.add(this);
return this;
}
}
public class Builder
{
int[] from;
int[] to;
int[] verbs;
private Builder(int[] verbs)
{
this.verbs = verbs;
}
public Builder from(int ... nums)
{
from = nums;
return this;
}
public Builder to(int ... nums)
{
to = nums;
return this;
}
public Filter ready()
{
return new Filter(from, to, verbs);
}
public Filter drop()
{
return ready().drop();
}
}
public Builder verbs(MessagingService.Verb ... verbs)
{
int[] ids = new int[verbs.length];
for (int i = 0 ; i < verbs.length ; ++i)
ids[i] = verbs[i].ordinal();
return new Builder(ids);
}
public Builder allVerbs()
{
return new Builder(null);
}
public void reset()
{
filters.clear();
}
}

47
test/distributed/org/apache/cassandra/distributed/RowUtil.java Normal file
View File

@ -0,0 +1,47 @@
/*
* 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.List;
import org.apache.cassandra.cql3.ColumnSpecification;
import org.apache.cassandra.transport.messages.ResultMessage;
public class RowUtil
{
public static Object[][] toObjects(ResultMessage.Rows rows)
{
Object[][] result = new Object[rows.result.rows.size()][];
List<ColumnSpecification> specs = rows.result.metadata.names;
for (int i = 0; i < rows.result.rows.size(); i++)
{
List<ByteBuffer> row = rows.result.rows.get(i);
result[i] = new Object[row.size()];
for (int j = 0; j < row.size(); j++)
{
ByteBuffer bb = row.get(j);
if (bb != null)
result[i][j] = specs.get(j).type.getSerializer().deserialize(bb);
}
}
return result;
}
}

277
test/distributed/org/apache/cassandra/distributed/TestCluster.java Normal file
View File

@ -0,0 +1,277 @@
/*
* 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.ExecutorService;
import java.util.concurrent.Executors;
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.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
{
private static ExecutorService exec = Executors.newCachedThreadPool(new NamedThreadFactory("cluster-async-tasks"));
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 -> exec.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 -> exec.submit(i::shutdown))
.collect(Collectors.toList());
// withThreadLeakCheck(futures);
// Make sure to only delete directory when threads are stopped
exec.submit(() -> {
FBUtilities.waitOnFutures(futures);
FileUtils.deleteRecursive(root);
});
}
// 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));
}
}
}