From 1d5e0301f36b5c0dbd8fa8b5e599d2bb413f1b4e Mon Sep 17 00:00:00 2001 From: Alex Petrov Date: Fri, 16 Nov 2018 19:41:58 +0100 Subject: [PATCH 1/4] 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 --- .circleci/config.yml | 14 +- build.xml | 18 +- ide/idea-iml-file.xml | 1 + .../cassandra/auth/PermissionsCache.java | 14 +- .../org/apache/cassandra/auth/RolesCache.java | 11 +- .../concurrent/InfiniteLoopExecutor.java | 89 +++++ .../JMXEnabledThreadPoolExecutor.java | 24 +- .../concurrent/ScheduledExecutors.java | 15 + .../concurrent/SharedExecutorPool.java | 15 +- .../cassandra/concurrent/StageManager.java | 10 + .../org/apache/cassandra/config/Config.java | 16 +- .../cassandra/config/DatabaseDescriptor.java | 6 +- .../apache/cassandra/cql3/QueryOptions.java | 7 +- .../apache/cassandra/cql3/QueryProcessor.java | 8 +- .../cassandra/db/BlacklistedDirectories.java | 17 +- .../cassandra/db/ColumnFamilyStore.java | 36 +- .../cassandra/db/HintedHandOffManager.java | 16 +- .../org/apache/cassandra/db/Memtable.java | 3 +- .../cassandra/db/commitlog/CommitLog.java | 15 +- .../db/compaction/CompactionManager.java | 20 +- .../apache/cassandra/gms/FailureDetector.java | 14 +- .../org/apache/cassandra/gms/Gossiper.java | 15 +- .../io/sstable/IndexSummaryManager.java | 26 +- .../locator/DynamicEndpointSnitch.java | 24 +- .../cassandra/locator/EndpointSnitchInfo.java | 14 +- .../cassandra/locator/InetAddressAndPort.java | 194 +++++++++ .../metrics/CassandraMetricsRegistry.java | 11 +- .../cassandra/net/MessagingService.java | 13 +- .../cassandra/service/CacheService.java | 15 +- .../cassandra/service/CassandraDaemon.java | 11 +- .../PendingRangeCalculatorService.java | 9 + .../cassandra/service/StorageProxy.java | 14 +- .../cassandra/service/StorageService.java | 16 +- .../cassandra/utils/ByteBufferUtil.java | 31 ++ .../apache/cassandra/utils/MBeanWrapper.java | 209 ++++++++++ .../org/apache/cassandra/utils/Mx4jTool.java | 4 +- .../cassandra/utils/concurrent/Ref.java | 47 ++- .../utils/memory/MemtableCleanerThread.java | 85 ++-- .../cassandra/utils/memory/MemtablePool.java | 7 + .../jmx/LegacyJMXProgressSupport.java | 4 +- test/conf/logback-dtest.xml | 79 ++++ .../cassandra/distributed/Coordinator.java | 79 ++++ .../DistributedReadWritePathTest.java | 92 +++++ .../distributed/DistributedTestBase.java | 86 ++++ .../cassandra/distributed/Instance.java | 371 ++++++++++++++++++ .../distributed/InstanceClassLoader.java | 101 +++++ .../cassandra/distributed/InstanceConfig.java | 87 ++++ .../distributed/InstanceIDDefiner.java | 38 ++ .../distributed/InvokableInstance.java | 133 +++++++ .../cassandra/distributed/LegacyAdapter.java | 42 ++ .../apache/cassandra/distributed/Message.java | 41 ++ .../cassandra/distributed/MessageFilters.java | 175 +++++++++ .../apache/cassandra/distributed/RowUtil.java | 47 +++ .../cassandra/distributed/TestCluster.java | 277 +++++++++++++ 54 files changed, 2437 insertions(+), 329 deletions(-) create mode 100644 src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java create mode 100644 src/java/org/apache/cassandra/locator/InetAddressAndPort.java create mode 100644 src/java/org/apache/cassandra/utils/MBeanWrapper.java create mode 100644 test/conf/logback-dtest.xml create mode 100644 test/distributed/org/apache/cassandra/distributed/Coordinator.java create mode 100644 test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java create mode 100644 test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java create mode 100644 test/distributed/org/apache/cassandra/distributed/Instance.java create mode 100644 test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java create mode 100644 test/distributed/org/apache/cassandra/distributed/InstanceConfig.java create mode 100644 test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java create mode 100644 test/distributed/org/apache/cassandra/distributed/InvokableInstance.java create mode 100644 test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java create mode 100644 test/distributed/org/apache/cassandra/distributed/Message.java create mode 100644 test/distributed/org/apache/cassandra/distributed/MessageFilters.java create mode 100644 test/distributed/org/apache/cassandra/distributed/RowUtil.java create mode 100644 test/distributed/org/apache/cassandra/distributed/TestCluster.java diff --git a/.circleci/config.yml b/.circleci/config.yml index 44707e21b1..d10c0d8a01 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -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/ diff --git a/build.xml b/build.xml index 7c250691b1..76918235db 100644 --- a/build.xml +++ b/build.xml @@ -57,6 +57,7 @@ + @@ -65,11 +66,14 @@ + + + @@ -97,7 +101,8 @@ - + + @@ -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}"> @@ -1195,6 +1200,7 @@ + @@ -1333,7 +1339,7 @@ - + @@ -1405,6 +1411,7 @@ + @@ -1767,7 +1774,7 @@ e.g. org/apache/cassandra/hints/HintMessageTest.java --> - + @@ -1837,6 +1844,7 @@ + diff --git a/ide/idea-iml-file.xml b/ide/idea-iml-file.xml index 63d0e1d5ce..0f2af9b942 100644 --- a/ide/idea-iml-file.xml +++ b/ide/idea-iml-file.xml @@ -33,6 +33,7 @@ + diff --git a/src/java/org/apache/cassandra/auth/PermissionsCache.java b/src/java/org/apache/cassandra/auth/PermissionsCache.java index 8746b36e5d..c8d777e12f 100644 --- a/src/java/org/apache/cassandra/auth/PermissionsCache.java +++ b/src/java/org/apache/cassandra/auth/PermissionsCache.java @@ -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 getPermissions(AuthenticatedUser user, IResource resource) diff --git a/src/java/org/apache/cassandra/auth/RolesCache.java b/src/java/org/apache/cassandra/auth/RolesCache.java index 554df9e854..75ac89da28 100644 --- a/src/java/org/apache/cassandra/auth/RolesCache.java +++ b/src/java/org/apache/cassandra/auth/RolesCache.java @@ -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 getRoles(RoleResource role) diff --git a/src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java b/src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java new file mode 100644 index 0000000000..3b73b0d90a --- /dev/null +++ b/src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java @@ -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)); + } +} diff --git a/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java index 2b8670155c..377442bfc4 100644 --- a/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java @@ -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 ScheduledThreadPoolExecutor. 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(); diff --git a/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java b/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java index 5962db9944..60f07dc744 100644 --- a/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java +++ b/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java @@ -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); + } } diff --git a/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java b/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java index dfd701128a..8b53dfb366 100644 --- a/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java +++ b/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java @@ -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 executors = new CopyOnWriteArrayList<>(); + public final List 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); + } } diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java index 343648c07b..5e0a66770c 100644 --- a/src/java/org/apache/cassandra/concurrent/StageManager.java +++ b/src/java/org/apache/cassandra/concurrent/StageManager.java @@ -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 diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 64b36a0dfc..7836396a6a 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -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 overrideLoadConfig = null; public Integer file_cache_size_in_mb; @@ -313,6 +311,16 @@ public class Config isClientMode = clientMode; } + public static Supplier getOverrideLoadConfig() + { + return overrideLoadConfig; + } + + public static void setOverrideLoadConfig(Supplier loadConfig) + { + overrideLoadConfig = loadConfig; + } + public void configHintedHandoff() throws ConfigurationException { if (hinted_handoff_enabled != null && !hinted_handoff_enabled.isEmpty()) diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 90a82fe831..d1a5c66b79 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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.construct(loaderClass, "configuration loading"); Config config = loader.loadConfig(); diff --git a/src/java/org/apache/cassandra/cql3/QueryOptions.java b/src/java/org/apache/cassandra/cql3/QueryOptions.java index be773e16c7..da705e0f4a 100644 --- a/src/java/org/apache/cassandra/cql3/QueryOptions.java +++ b/src/java/org/apache/cassandra/cql3/QueryOptions.java @@ -83,7 +83,12 @@ public abstract class QueryOptions public static QueryOptions create(ConsistencyLevel consistency, List 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 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 columnSpecs) diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java index c70267963c..a0afda7242 100644 --- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java @@ -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 boundValues = new ArrayList(values.length); + List boundValues = new ArrayList<>(values.length); for (int i = 0; i < values.length; i++) { Object value = values[i]; diff --git a/src/java/org/apache/cassandra/db/BlacklistedDirectories.java b/src/java/org/apache/cassandra/db/BlacklistedDirectories.java index f47fd572a6..a14e0139f5 100644 --- a/src/java/org/apache/cassandra/db/BlacklistedDirectories.java +++ b/src/java/org/apache/cassandra/db/BlacklistedDirectories.java @@ -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 getUnreadableDirectories() diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 45908deb6a..d26cd610bf 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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(); } diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index 74f0a72e30..95af9baa55 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -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() diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index b4ada096be..b17b3fc7b2 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -52,7 +52,8 @@ public class Memtable implements Comparable { 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; diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index 460ecfedb6..2f0179d1ab 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -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(); } diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index c37049f270..7871b177f3 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -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> ranges) { this.ranges = ranges; - cacheCleanupExecutor.submit(new Runnable() + instance.cacheCleanupExecutor.submit(new Runnable() { @Override public void run() diff --git a/src/java/org/apache/cassandra/gms/FailureDetector.java b/src/java/org/apache/cassandra/gms/FailureDetector.java index b9b79446ca..679d0b8e65 100644 --- a/src/java/org/apache/cassandra/gms/FailureDetector.java +++ b/src/java/org/apache/cassandra/gms/FailureDetector.java @@ -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() diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index a68f4eae19..831c252254 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -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) diff --git a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java index 0e9073f22e..586e4f2f0e 100644 --- a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java +++ b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java @@ -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() diff --git a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java index 9c0c57ebf8..d6a601c7ec 100644 --- a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java +++ b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java @@ -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 diff --git a/src/java/org/apache/cassandra/locator/EndpointSnitchInfo.java b/src/java/org/apache/cassandra/locator/EndpointSnitchInfo.java index bbfabb6718..be28b3c8e4 100644 --- a/src/java/org/apache/cassandra/locator/EndpointSnitchInfo.java +++ b/src/java/org/apache/cassandra/locator/EndpointSnitchInfo.java @@ -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 diff --git a/src/java/org/apache/cassandra/locator/InetAddressAndPort.java b/src/java/org/apache/cassandra/locator/InetAddressAndPort.java new file mode 100644 index 0000000000..6daa2e132c --- /dev/null +++ b/src/java/org/apache/cassandra/locator/InetAddressAndPort.java @@ -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, 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; + } +} diff --git a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java index 8e5671b581..d525a26260 100644 --- a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java +++ b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java @@ -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 diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index f0e2fbfb3f..85cd51c7a9 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -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); } } diff --git a/src/java/org/apache/cassandra/service/CacheService.java b/src/java/org/apache/cassandra/service/CacheService.java index a13a52d3f2..3872e5f843 100644 --- a/src/java/org/apache/cassandra/service/CacheService.java +++ b/src/java/org/apache/cassandra/service/CacheService.java @@ -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(); diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index e250050d1e..8f6c9c21f9 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -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()) { diff --git a/src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java b/src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java index 116cede3c8..e82b0bbd15 100644 --- a/src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java +++ b/src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java @@ -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); + } } diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 7b7979df26..b734343e61 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -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() { diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 03470b6a24..3f87f708b2 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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 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); diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java index 6c676e0051..1779e67639 100644 --- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java +++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java @@ -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); diff --git a/src/java/org/apache/cassandra/utils/MBeanWrapper.java b/src/java/org/apache/cassandra/utils/MBeanWrapper.java new file mode 100644 index 0000000000..3b5c7cb70c --- /dev/null +++ b/src/java/org/apache/cassandra/utils/MBeanWrapper.java @@ -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() + { + public void accept(Exception e) + { + throw new RuntimeException(e); + } + }), + LOG(new Consumer() + { + public void accept(Exception e) + { + logger.error("Error in MBean wrapper: ", e); + } + }), + IGNORE(new Consumer() + { + public void accept(Exception e) + { + + } + }); + + private Consumer handler; + OnException(Consumer handler) + { + this.handler = handler; + } + } +} diff --git a/src/java/org/apache/cassandra/utils/Mx4jTool.java b/src/java/org/apache/cassandra/utils/Mx4jTool.java index 41a5b332b3..77a6013522 100644 --- a/src/java/org/apache/cassandra/utils/Mx4jTool.java +++ b/src/java/org/apache/cassandra/utils/Mx4jTool.java @@ -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"); diff --git a/src/java/org/apache/cassandra/utils/concurrent/Ref.java b/src/java/org/apache/cassandra/utils/concurrent/Ref.java index 65fe9a844f..e1cc7ff6a0 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Ref.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Ref.java @@ -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 implements RefCounted private static final Set globallyExtant = Collections.newSetFromMap(new ConcurrentHashMap()); static final ReferenceQueue 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); } } diff --git a/src/java/org/apache/cassandra/utils/memory/MemtableCleanerThread.java b/src/java/org/apache/cassandra/utils/memory/MemtableCleanerThread.java index 5a90463530..628b8c00c9 100644 --- a/src/java/org/apache/cassandra/utils/memory/MemtableCleanerThread.java +++ b/src/java/org/apache/cassandra/utils/memory/MemtableCleanerThread.java @@ -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

extends Thread +public class MemtableCleanerThread

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

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

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(); + } } diff --git a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java index bb858843b9..a096ec3a55 100644 --- a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java +++ b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java @@ -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(); /** diff --git a/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java b/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java index fae6f2a2ea..438e4116c3 100644 --- a/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java +++ b/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java @@ -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; diff --git a/test/conf/logback-dtest.xml b/test/conf/logback-dtest.xml new file mode 100644 index 0000000000..b8019f6d0b --- /dev/null +++ b/test/conf/logback-dtest.xml @@ -0,0 +1,79 @@ + + + + + + + + + + + + + + ./build/test/logs/${cassandra.testtag}/TEST-${suitename}.log + + ./build/test/logs/${cassandra.testtag}/TEST-${suitename}.log.%i.gz + 1 + 20 + + + + 20MB + + + + %-5level [%thread] ${instance_id} %date{ISO8601} %msg%n + false + + + + + + %-5level [%thread] ${instance_id} %date{ISO8601} %F:%L - %msg%n + + + DEBUG + + + + + + + + + + + + + + + 0 + 0 + 1024 + + true + + + + + + diff --git a/test/distributed/org/apache/cassandra/distributed/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/Coordinator.java new file mode 100644 index 0000000000..5cb51a4f7c --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/Coordinator.java @@ -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 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); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java b/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java new file mode 100644 index 0000000000..ad0cacc58e --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java @@ -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)); + } + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java b/test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java new file mode 100644 index 0000000000..64ef8592ae --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java @@ -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; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/Instance.java b/test/distributed/org/apache/cassandra/distributed/Instance.java new file mode 100644 index 0000000000..7564476051 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/Instance.java @@ -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 deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message); + BiConsumer deliverToInstanceIfNotFiltered = cluster.filters().filter(deliverToInstance); + + Map addressAndPortMap = new HashMap<>(); + cluster.stream().map(Instance::getBroadcastAddress).forEach(addressAndPort -> { + if (null != addressAndPortMap.put(addressAndPort.address, addressAndPort)) + throw new IllegalStateException("This version of Cassandra does not support multiple nodes with the same InetAddress"); + }); + + acceptsOnInstance((BiConsumer deliver) -> + MessagingService.instance().addMessageSink(new MessageDeliverySink(deliver, addressAndPortMap::get)) + ).accept(deliverToInstanceIfNotFiltered); + } + + private static class MessageDeliverySink implements IMessageSink + { + private final BiConsumer deliver; + private final Function lookupAddressAndPort; + MessageDeliverySink(BiConsumer deliver, Function 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 initialTokens = new ArrayList<>(); + List hosts = new ArrayList<>(); + List hostIds = new ArrayList<>(); + for (int i = 1 ; i <= cluster.size() ; ++i) + { + InstanceConfig config = cluster.get(i).config; + initialTokens.add(config.initial_token); + try + { + hosts.add(InetAddressAndPort.getByName(config.broadcast_address)); + } + catch (UnknownHostException e) + { + throw new RuntimeException(e); + } + hostIds.add(config.hostId); + } + + runOnInstance(() -> + { + try + { + IPartitioner partitioner = FBUtilities.newPartitioner(partitionerName); + StorageService storageService = StorageService.instance; + List tokens = new ArrayList<>(); + for (String token : initialTokens) + tokens.add(partitioner.getTokenFactory().fromString(token)); + + for (int i = 0; i < tokens.size(); i++) + { + InetAddressAndPort ep = hosts.get(i); + Gossiper.instance.initializeNodeUnsafe(ep.address, hostIds.get(i), 1); + Gossiper.instance.injectApplicationState(ep.address, + ApplicationState.TOKENS, + new VersionedValue.VersionedValueFactory(partitioner).tokens(Collections.singleton(tokens.get(i)))); + storageService.onChange(ep.address, + ApplicationState.STATUS, + new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(tokens.get(i)))); + Gossiper.instance.realMarkAlive(ep.address, Gossiper.instance.getEndpointStateForEndpoint(ep.address)); + MessagingService.instance().setVersion(ep.address, MessagingService.current_version); + } + + // check that all nodes are in token metadata + for (int i = 0; i < tokens.size(); ++i) + assert storageService.getTokenMetadata().isMember(hosts.get(i).address); + } + catch (Throwable e) // UnknownHostException + { + 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; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java b/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java new file mode 100644 index 0000000000..036f6b1ee4 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java @@ -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 isCommonClassName; + + InstanceClassLoader(int id, URL[] urls, Predicate isCommonClassName, ClassLoader commonClassLoader) + { + super(urls, null); + this.id = id; + this.commonClassLoader = commonClassLoader; + this.isCommonClassName = isCommonClassName; + } + + @Override + public Class loadClass(String name) throws ClassNotFoundException + { + // Do not share: + // * yaml, which is a rare exception because it does mess with loading org.cassandra...Config class instances + // * most of the rest of Cassandra classes (unless they were explicitly shared) g + if (name.startsWith("org.slf4j") || + name.startsWith("ch.qos.logback") || + name.startsWith("org.yaml") || + (name.startsWith("org.apache.cassandra") && !isCommonClassName.apply(name))) + return loadClassInternal(name); + + return commonClassLoader.loadClass(name); + } + + Class loadClassInternal(String name) throws ClassNotFoundException + { + synchronized (getClassLoadingLock(name)) + { + // First, check if the class has already been loaded + Class c = findLoadedClass(name); + + if (c == null) + c = findClass(name); + + return c; + } + } + + public static IntFunction createFactory(URLClassLoader contextClassLoader) + { + Set commonClassNames = new HashSet<>(); + for (Class k : commonClasses) + commonClassNames.add(k.getName()); + + URL[] urls = contextClassLoader.getURLs(); + return id -> new InstanceClassLoader(id, urls, commonClassNames::contains, contextClassLoader); + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/InstanceConfig.java new file mode 100644 index 0000000000..49c2e1f624 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/InstanceConfig.java @@ -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); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java b/test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java new file mode 100644 index 0000000000..11677487fa --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java @@ -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 "

"; + else + return "INSTANCE" + instanceId; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java b/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java new file mode 100644 index 0000000000..f646ae1f8f --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java @@ -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 extends Callable, Serializable { public T call(); } + public SerializableCallable callsOnInstance(SerializableCallable call) { return (SerializableCallable) transferOneObject(call); } + public T callOnInstance(SerializableCallable 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 extends Consumer, Serializable {} + public SerializableConsumer acceptsOnInstance(SerializableConsumer consumer) { return (SerializableConsumer) transferOneObject(consumer); } + + public interface SerializableBiConsumer extends BiConsumer, Serializable {} + public SerializableBiConsumer acceptsOnInstance(SerializableBiConsumer consumer) { return (SerializableBiConsumer) transferOneObject(consumer); } + + public interface SerializableFunction extends Function, Serializable {} + public SerializableFunction appliesOnInstance(SerializableFunction f) { return (SerializableFunction) transferOneObject(f); } + + public interface SerializableBiFunction extends BiFunction, Serializable {} + public SerializableBiFunction appliesOnInstance(SerializableBiFunction f) { return (SerializableBiFunction) transferOneObject(f); } + + public interface SerializableTriFunction extends Serializable + { + O apply(I1 i1, I2 i2, I3 i3); + } + public SerializableTriFunction appliesOnInstance(SerializableTriFunction f) { return (SerializableTriFunction) transferOneObject(f); } + + public interface InstanceFunction extends SerializableBiFunction {} + + // E must be a functional interface, and lambda must be implemented by a lambda function + public E invokesOnInstance(E lambda) + { + return (E) transferOneObject(lambda); + } + + public Object transferOneObject(Object object) + { + byte[] bytes = serializeOneObject(object); + try + { + Object onInstance = deserializeOnInstance.invoke(null, bytes); + if (onInstance.getClass().getClassLoader() != classLoader) + throw new IllegalStateException(onInstance + " seemingly from wrong class loader: " + onInstance.getClass().getClassLoader() + ", but expected " + classLoader); + + return onInstance; + } + catch (IllegalAccessException | InvocationTargetException e) + { + throw new RuntimeException(e); + } + } + + private byte[] serializeOneObject(Object object) + { + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos)) + { + oos.writeObject(object); + oos.close(); + return baos.toByteArray(); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + @SuppressWarnings("unused") // called through method invocation + public static Object deserializeOneObject(byte[] bytes) + { + try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes); + ObjectInputStream ois = new ObjectInputStream(bais);) + { + return ois.readObject(); + } + catch (IOException | ClassNotFoundException e) + { + throw new RuntimeException(e); + } + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java b/test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java new file mode 100644 index 0000000000..1ff88eda75 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java @@ -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; + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/Message.java b/test/distributed/org/apache/cassandra/distributed/Message.java new file mode 100644 index 0000000000..b5492a2a2c --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/Message.java @@ -0,0 +1,41 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed; + +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; + } +} + diff --git a/test/distributed/org/apache/cassandra/distributed/MessageFilters.java b/test/distributed/org/apache/cassandra/distributed/MessageFilters.java new file mode 100644 index 0000000000..47eb52bcbd --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/MessageFilters.java @@ -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 filters = new CopyOnWriteArraySet<>(); + + public MessageFilters(TestCluster cluster) + { + this.cluster = cluster; + } + + BiConsumer filter(BiConsumer applyIfNotFiltered) + { + return (toAddress, message) -> + { + int from = cluster.get(message.from).config.num; + int to = cluster.get(toAddress).config.num; + int verb = message.verb; + 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(); + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/RowUtil.java b/test/distributed/org/apache/cassandra/distributed/RowUtil.java new file mode 100644 index 0000000000..bce896d784 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/RowUtil.java @@ -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 specs = rows.result.metadata.names; + for (int i = 0; i < rows.result.rows.size(); i++) + { + List 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; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/TestCluster.java b/test/distributed/org/apache/cassandra/distributed/TestCluster.java new file mode 100644 index 0000000000..20a4ab259a --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/TestCluster.java @@ -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 instances; + private final Coordinator coordinator; + private final Map instanceMap; + private final MessageFilters filters; + + private TestCluster(File root, List instances) + { + this.root = root; + this.instances = instances; + this.instanceMap = new HashMap<>(); + this.coordinator = new Coordinator(instances.get(0)); + this.filters = new MessageFilters(this); + } + + void launch() + { + FBUtilities.waitOnFutures(instances.stream() + .map(i -> 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 stream() { return instances.stream(); } + + public Instance get(InetAddressAndPort addr) + { + return instanceMap.get(addr); + } + + MessageFilters filters() + { + return filters; + } + + MessageFilters.Builder verbs(MessagingService.Verb ... verbs) + { + return filters.verbs(verbs); + } + + public void disableAutoCompaction(String keyspace) + { + for (Instance instance : instances) + { + instance.runOnInstance(() -> { + for (ColumnFamilyStore cs : Keyspace.open(keyspace).getColumnFamilyStores()) + cs.disableAutoCompaction(); + }); + } + } + + public void schemaChange(String query) + { + try (SchemaChangeMonitor monitor = new SchemaChangeMonitor()) + { + // execute the schema change + coordinator().execute(query, ConsistencyLevel.ALL); + monitor.waitForAgreement(); + } + } + + /** + * Will wait for a schema change AND agreement that occurs after it is created + * (and precedes the invocation to waitForAgreement) + * + * Works by simply checking if all UUIDs agree after any schema version change event, + * so long as the waitForAgreement method has been entered (indicating the change has + * taken place on the coordinator) + * + * This could perhaps be made a little more robust, but this should more than suffice. + */ + public class SchemaChangeMonitor implements AutoCloseable + { + public SchemaChangeMonitor() {} + + @Override + public void close() { } + + public void waitForAgreement() + { + long start = System.nanoTime(); + while (1 != instances.stream().map(Instance::getSchemaVersion).distinct().count()) + { + if (System.nanoTime() - start > TimeUnit.MINUTES.toNanos(1L)) + throw new IllegalStateException("Schema agreement not reached"); + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1L)); + } + } + } + + public void schemaChange(String statement, int instance) + { + get(instance).schemaChange(statement); + } + + public static TestCluster create(int nodeCount) throws Throwable + { + return create(nodeCount, Files.createTempDirectory("dtests").toFile()); + } + + public static TestCluster create(int nodeCount, File root) + { + root.mkdirs(); + setupLogging(root); + + IntFunction classLoaderFactory = InstanceClassLoader.createFactory( + (URLClassLoader) Thread.currentThread().getContextClassLoader()); + List instances = new ArrayList<>(); + long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount); + for (int i = 0 ; i < nodeCount ; ++i) + { + InstanceConfig instanceConfig = InstanceConfig.generate(i + 1, root, String.valueOf(token)); + instances.add(new Instance(instanceConfig, classLoaderFactory.apply(i + 1))); + token += increment; + } + + TestCluster cluster = new TestCluster(root, instances); + cluster.launch(); + return cluster; + } + + private static void setupLogging(File root) + { + try + { + String testConfPath = "test/conf/logback-dtest.xml"; + Path logConfPath = Paths.get(root.getPath(), "/logback-dtest.xml"); + if (!logConfPath.toFile().exists()) + { + Files.copy(new File(testConfPath).toPath(), + logConfPath); + } + System.setProperty("logback.configurationFile", "file://" + logConfPath); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + @Override + public void close() + { + List> futures = instances.stream() + .map(i -> 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> futures) + { + FBUtilities.waitOnFutures(futures); + + Set threadSet = Thread.getAllStackTraces().keySet(); + threadSet = Sets.difference(threadSet, Collections.singletonMap(Thread.currentThread(), null).keySet()); + if (!threadSet.isEmpty()) + { + for (Thread thread : threadSet) + { + System.out.println(thread); + System.out.println(Arrays.toString(thread.getStackTrace())); + } + throw new RuntimeException(String.format("Not all threads have shut down. %d threads are still running: %s", threadSet.size(), threadSet)); + } + } + +} + From a3004ef29fe223186b518a68e07a180e79939193 Mon Sep 17 00:00:00 2001 From: Alex Petrov Date: Thu, 11 Oct 2018 18:23:20 +0200 Subject: [PATCH 2/4] Make sure that SEP Worker shuts down when pool is shutdown Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-14815 --- .../cassandra/concurrent/SEPWorker.java | 17 ++++- .../concurrent/SharedExecutorPool.java | 35 +++++++--- .../cassandra/distributed/Instance.java | 2 +- .../cassandra/concurrent/SEPExecutorTest.java | 70 +++++++++++++++++++ 4 files changed, 113 insertions(+), 11 deletions(-) create mode 100644 test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java diff --git a/src/java/org/apache/cassandra/concurrent/SEPWorker.java b/src/java/org/apache/cassandra/concurrent/SEPWorker.java index 3b3e7ad9f5..edc31daa8c 100644 --- a/src/java/org/apache/cassandra/concurrent/SEPWorker.java +++ b/src/java/org/apache/cassandra/concurrent/SEPWorker.java @@ -71,9 +71,14 @@ final class SEPWorker extends AtomicReference implements Runnabl { while (true) { + if (pool.shuttingDown) + return; + if (isSpinning() && !selfAssign()) { doWaitSpin(); + // if the pool is terminating, but we have been assigned STOP_SIGNALLED, if we do not re-check + // whether the pool is shutting down this thread will go to sleep and block forever continue; } @@ -113,8 +118,12 @@ final class SEPWorker extends AtomicReference implements Runnabl // return our work permit, and maybe signal shutdown assigned.returnWorkPermit(); - if (shutdown && assigned.getActiveCount() == 0) - assigned.shutdown.signalAll(); + if (shutdown) + { + if (assigned.getActiveCount() == 0) + assigned.shutdown.signalAll(); + return; + } assigned = null; // try to immediately reassign ourselves some work; if we fail, start spinning @@ -164,7 +173,11 @@ final class SEPWorker extends AtomicReference implements Runnabl // if we're being descheduled, place ourselves in the descheduled collection if (work.isStop()) + { pool.descheduled.put(workerId, this); + if (pool.shuttingDown) + return true; + } // if we're currently stopped, and the new state is not a stop signal // (which we can immediately convert to stopped), unpark the worker diff --git a/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java b/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java index 8b53dfb366..d355d77b4b 100644 --- a/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java +++ b/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java @@ -24,8 +24,7 @@ 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 java.util.concurrent.locks.LockSupport; import static org.apache.cassandra.concurrent.SEPWorker.Work; @@ -77,6 +76,8 @@ public class SharedExecutorPool // the collection of threads that have been asked to stop/deschedule - new workers are scheduled from here last final ConcurrentSkipListMap descheduled = new ConcurrentSkipListMap<>(); + volatile boolean shuttingDown = false; + public SharedExecutorPool(String poolName) { this.poolName = poolName; @@ -113,13 +114,31 @@ public class SharedExecutorPool return executor; } - @VisibleForTesting - public static void shutdownSharedPool() throws InterruptedException + public void shutdown() throws InterruptedException { - for (SEPExecutor executor : SHARED.executors) - executor.shutdown(); + shuttingDown = true; + for (SEPExecutor executor : executors) + executor.shutdownNow(); - for (SEPExecutor executor : SHARED.executors) - executor.awaitTermination(60, TimeUnit.SECONDS); + terminateWorkers(); + + long until = System.nanoTime() + TimeUnit.MINUTES.toNanos(1L); + for (SEPExecutor executor : executors) + executor.shutdown.await(until - System.nanoTime(), TimeUnit.NANOSECONDS); + } + + void terminateWorkers() + { + assert shuttingDown; + + // To terminate our workers, we only need to unpark thread to make it runnable again, + // so that the pool.shuttingDown boolean is checked. If work was already in the process + // of being scheduled, worker will terminate upon running the task. + Map.Entry e; + while (null != (e = descheduled.pollFirstEntry())) + e.getValue().assign(Work.SPINNING, false); + + while (null != (e = spinning.pollFirstEntry())) + LockSupport.unpark(e.getValue().thread); } } diff --git a/test/distributed/org/apache/cassandra/distributed/Instance.java b/test/distributed/org/apache/cassandra/distributed/Instance.java index 7564476051..cd5a002529 100644 --- a/test/distributed/org/apache/cassandra/distributed/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/Instance.java @@ -341,8 +341,8 @@ public class Instance extends InvokableInstance Ref::shutdownReferenceReaper, Memtable.MEMORY_POOL::shutdown, StageManager::shutdownAndWait, - SharedExecutorPool::shutdownSharedPool, MessagingService.instance()::shutdown, + SharedExecutorPool.SHARED::shutdown, ScheduledExecutors::shutdownAndWait); Throwables.maybeFail(error); }); diff --git a/test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java b/test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java new file mode 100644 index 0000000000..011a8bac17 --- /dev/null +++ b/test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java @@ -0,0 +1,70 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.concurrent; + +import java.io.OutputStream; +import java.io.PrintStream; +import java.util.Arrays; +import java.util.concurrent.ExecutorService; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.cassandra.utils.FBUtilities; + +public class SEPExecutorTest +{ + @Test + public void shutdownTest() throws Throwable + { + for (int i = 0; i < 1000; i++) + { + shutdownOnce(i); + } + } + + private static void shutdownOnce(int run) throws Throwable + { + SharedExecutorPool sharedPool = new SharedExecutorPool("SharedPool"); + String MAGIC = "UNREPEATABLE_MAGIC_STRING"; + OutputStream nullOutputStream = new OutputStream() { + public void write(int b) { } + }; + PrintStream nullPrintSteam = new PrintStream(nullOutputStream); + + for (int idx = 0; idx < 20; idx++) + { + ExecutorService es = sharedPool.newExecutor(FBUtilities.getAvailableProcessors(), Integer.MAX_VALUE, "STAGE", run + MAGIC + idx); + // Write to black hole + es.execute(() -> nullPrintSteam.println("TEST" + es)); + } + + // shutdown does not guarantee that threads are actually dead once it exits, only that they will stop promptly afterwards + sharedPool.shutdown(); + for (Thread thread : Thread.getAllStackTraces().keySet()) + { + if (thread.getName().contains(MAGIC)) + { + thread.join(100); + if (thread.isAlive()) + Assert.fail(thread + " is still running " + Arrays.toString(thread.getStackTrace())); + } + } + } +} From 8ad4a9a522aef7243b2612d084bbae69ea8f1be6 Mon Sep 17 00:00:00 2001 From: Joseph Lynch Date: Fri, 7 Dec 2018 18:29:08 -0800 Subject: [PATCH 3/4] In JVM dtests need to clean up after instance shutdown Adds additional cleanup logic to ensure we don't leak classloaders and their associated objects when running the in JVM dtests. Patch by Joseph Lynch; reviewed by Alex Petrov for CASSANDRA-14922 --- build.xml | 7 +++ ide/idea-iml-file.xml | 3 +- ide/idea/workspace.xml | 4 +- .../concurrent/ScheduledExecutors.java | 2 +- .../io/sstable/format/SSTableReader.java | 7 +++ .../cassandra/net/MessagingService.java | 1 + .../cassandra/service/StorageService.java | 5 ++ .../utils/BackgroundActivityMonitor.java | 12 ++++ .../cassandra/utils/memory/MemtablePool.java | 4 ++ .../DistributedReadWritePathTest.java | 1 + .../cassandra/distributed/Instance.java | 11 +++- .../cassandra/distributed/TestCluster.java | 57 +++++++++++++++++-- 12 files changed, 104 insertions(+), 10 deletions(-) diff --git a/build.xml b/build.xml index 76918235db..8b524f7737 100644 --- a/build.xml +++ b/build.xml @@ -1246,6 +1246,13 @@ + + + diff --git a/ide/idea-iml-file.xml b/ide/idea-iml-file.xml index 0f2af9b942..338d7806e9 100644 --- a/ide/idea-iml-file.xml +++ b/ide/idea-iml-file.xml @@ -33,7 +33,8 @@ - + + diff --git a/ide/idea/workspace.xml b/ide/idea/workspace.xml index 1645d568c8..e632b5f040 100644 --- a/ide/idea/workspace.xml +++ b/ide/idea/workspace.xml @@ -168,7 +168,7 @@ - - - - - - - - - - 0 - 0 - 1024 - - true - - - + + + diff --git a/test/distributed/org/apache/cassandra/distributed/Instance.java b/test/distributed/org/apache/cassandra/distributed/Instance.java index 37be73c7b3..c8977d7c2a 100644 --- a/test/distributed/org/apache/cassandra/distributed/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/Instance.java @@ -30,6 +30,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; import java.util.function.BiConsumer; import java.util.function.Function; @@ -83,7 +85,7 @@ public class Instance extends InvokableInstance public Instance(InstanceConfig config, ClassLoader classLoader) { - super(classLoader); + super("node" + config.num, classLoader); this.config = config; } @@ -332,9 +334,9 @@ public class Instance extends InvokableInstance void shutdown() { - runOnInstance(() -> { + acceptsOnInstance((ExecutorService executor) -> { Throwable error = null; - error = runAndMergeThrowable(error, + error = parallelRun(error, executor, Gossiper.instance::stop, CompactionManager.instance::forceShutdown, CommitLog.instance::shutdownBlocking, @@ -345,32 +347,54 @@ public class Instance extends InvokableInstance StorageService.instance::shutdownBGMonitor, Ref::shutdownReferenceReaper, Memtable.MEMORY_POOL::shutdown, - StageManager::shutdownAndWait, - MessagingService.instance()::shutdown, - SharedExecutorPool.SHARED::shutdown, ScheduledExecutors::shutdownAndWait, - SSTableReader::shutdownBlocking); - + SSTableReader::shutdownBlocking + ); + error = parallelRun(error, executor, + MessagingService.instance()::shutdown + ); + error = parallelRun(error, executor, + StageManager::shutdownAndWait, + SharedExecutorPool.SHARED::shutdown + ); LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); loggerContext.stop(); Throwables.maybeFail(error); - }); + }).accept(isolatedExecutor); + super.shutdown(); } - private static Throwable runAndMergeThrowable(Throwable existing, ThrowingRunnable ... runnables) + private static Throwable parallelRun(Throwable accumulate, ExecutorService runOn, ThrowingRunnable ... runnables) { + List> results = new ArrayList<>(); for (ThrowingRunnable runnable : runnables) + { + results.add(runOn.submit(() -> { + try + { + runnable.run(); + return null; + } + catch (Throwable t) + { + return t; + } + })); + } + for (Future future : results) { try { - runnable.run(); + Throwable t = future.get(); + if (t != null) + throw t; } catch (Throwable t) { - existing = Throwables.merge(existing, t); + accumulate = Throwables.merge(accumulate, t); } } - return existing; + return accumulate; } public static interface ThrowingRunnable diff --git a/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java b/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java index 036f6b1ee4..9958fd3960 100644 --- a/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java +++ b/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java @@ -98,4 +98,12 @@ public class InstanceClassLoader extends URLClassLoader return id -> new InstanceClassLoader(id, urls, commonClassNames::contains, contextClassLoader); } + /** + * @return true iff this class was loaded by an InstanceClassLoader, and as such is used by a dtest node + */ + public static boolean wasLoadedByAnInstanceClassLoader(Class clazz) + { + return clazz.getClassLoader().getClass().getName().equals(InstanceClassLoader.class.getName()); + } + } diff --git a/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java b/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java index f646ae1f8f..9fb543d288 100644 --- a/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java +++ b/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java @@ -20,6 +20,7 @@ package org.apache.cassandra.distributed; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.Closeable; import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; @@ -27,18 +28,26 @@ import java.io.Serializable; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; +import org.apache.cassandra.concurrent.NamedThreadFactory; +import org.apache.cassandra.utils.Throwables; + public abstract class InvokableInstance { + protected final ExecutorService isolatedExecutor; private final ClassLoader classLoader; private final Method deserializeOnInstance; - public InvokableInstance(ClassLoader classLoader) + public InvokableInstance(String name, ClassLoader classLoader) { + this.isolatedExecutor = Executors.newCachedThreadPool(new NamedThreadFactory(name, Thread.NORM_PRIORITY, classLoader, new ThreadGroup(name))); this.classLoader = classLoader; try { @@ -50,31 +59,34 @@ public abstract class InvokableInstance } } - public interface SerializableCallable extends Callable, Serializable { public T call(); } - public SerializableCallable callsOnInstance(SerializableCallable call) { return (SerializableCallable) transferOneObject(call); } + public interface CallableNoExcept extends Callable { public T call(); } + public interface SerializableCallable extends CallableNoExcept, Serializable { } + public CallableNoExcept callsOnInstance(SerializableCallable call) { return invokesOnExecutor((SerializableCallable) transferOneObject(call), isolatedExecutor); } public T callOnInstance(SerializableCallable call) { return callsOnInstance(call).call(); } public interface SerializableRunnable extends Runnable, Serializable {} - public SerializableRunnable runsOnInstance(SerializableRunnable run) { return (SerializableRunnable) transferOneObject(run); } + public Runnable runsOnInstance(SerializableRunnable run) { return invokesOnExecutor((SerializableRunnable) transferOneObject(run), isolatedExecutor); } public void runOnInstance(SerializableRunnable run) { runsOnInstance(run).run(); } public interface SerializableConsumer extends Consumer, Serializable {} - public SerializableConsumer acceptsOnInstance(SerializableConsumer consumer) { return (SerializableConsumer) transferOneObject(consumer); } + public Consumer acceptsOnInstance(SerializableConsumer consumer) { return invokesOnExecutor((SerializableConsumer) transferOneObject(consumer), isolatedExecutor); } public interface SerializableBiConsumer extends BiConsumer, Serializable {} - public SerializableBiConsumer acceptsOnInstance(SerializableBiConsumer consumer) { return (SerializableBiConsumer) transferOneObject(consumer); } + public BiConsumer acceptsOnInstance(SerializableBiConsumer consumer) { return invokesOnExecutor((SerializableBiConsumer) transferOneObject(consumer), isolatedExecutor); } public interface SerializableFunction extends Function, Serializable {} - public SerializableFunction appliesOnInstance(SerializableFunction f) { return (SerializableFunction) transferOneObject(f); } + public Function appliesOnInstance(SerializableFunction f) { return invokesOnExecutor((SerializableFunction) transferOneObject(f), isolatedExecutor); } public interface SerializableBiFunction extends BiFunction, Serializable {} - public SerializableBiFunction appliesOnInstance(SerializableBiFunction f) { return (SerializableBiFunction) transferOneObject(f); } + public BiFunction appliesOnInstance(SerializableBiFunction f) { return invokesOnExecutor((SerializableBiFunction) transferOneObject(f), isolatedExecutor); } - public interface SerializableTriFunction extends Serializable + public interface TriFunction { O apply(I1 i1, I2 i2, I3 i3); } - public SerializableTriFunction appliesOnInstance(SerializableTriFunction f) { return (SerializableTriFunction) transferOneObject(f); } + public interface SerializableTriFunction extends Serializable, TriFunction { } + + public TriFunction appliesOnInstance(SerializableTriFunction f) { return invokesOnExecutor((SerializableTriFunction) transferOneObject(f), isolatedExecutor); } public interface InstanceFunction extends SerializableBiFunction {} @@ -130,4 +142,72 @@ public abstract class InvokableInstance } } + private static CallableNoExcept invokesOnExecutor(SerializableCallable callable, ExecutorService invokeOn) + { + return () -> { + try + { + return invokeOn.submit(callable).get(); + } + catch (InterruptedException e) + { + throw new RuntimeException(e); + } + catch (ExecutionException e) + { + Throwables.maybeFail(e.getCause()); + throw new AssertionError(); + } + }; + } + + private static Runnable invokesOnExecutor(SerializableRunnable runnable, ExecutorService invokeOn) + { + return () -> { + try + { + invokeOn.submit(runnable).get(); + } + catch (InterruptedException e) + { + throw new RuntimeException(e); + } + catch (ExecutionException e) + { + Throwables.maybeFail(e.getCause()); + throw new AssertionError(); + } + }; + } + + private static Consumer invokesOnExecutor(SerializableConsumer consumer, ExecutorService invokeOn) + { + return (a) -> invokesOnExecutor(() -> consumer.accept(a), invokeOn).run(); + } + + private static BiConsumer invokesOnExecutor(SerializableBiConsumer consumer, ExecutorService invokeOn) + { + return (a, b) -> invokesOnExecutor(() -> consumer.accept(a, b), invokeOn).run(); + } + + private static Function invokesOnExecutor(SerializableFunction f, ExecutorService invokeOn) + { + return (a) -> invokesOnExecutor(() -> f.apply(a), invokeOn).call(); + } + + private static BiFunction invokesOnExecutor(SerializableBiFunction f, ExecutorService invokeOn) + { + return (a, b) -> invokesOnExecutor(() -> f.apply(a, b), invokeOn).call(); + } + + private static SerializableTriFunction invokesOnExecutor(SerializableTriFunction f, ExecutorService invokeOn) + { + return (a, b, c) -> invokesOnExecutor(() -> f.apply(a, b, c), invokeOn).call(); + } + + void shutdown() + { + isolatedExecutor.shutdownNow(); + } + } diff --git a/test/distributed/org/apache/cassandra/distributed/TestCluster.java b/test/distributed/org/apache/cassandra/distributed/TestCluster.java index 82b775aa3a..245030641d 100644 --- a/test/distributed/org/apache/cassandra/distributed/TestCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/TestCluster.java @@ -31,24 +31,19 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.LockSupport; -import java.util.concurrent.TimeoutException; import java.util.function.IntFunction; import java.util.stream.Collectors; import java.util.stream.Stream; import com.google.common.collect.Sets; -import io.netty.util.concurrent.FastThreadLocal; -import io.netty.util.concurrent.FastThreadLocalThread; -import io.netty.util.internal.InternalThreadLocalMap; -import org.apache.cassandra.concurrent.NamedThreadFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.Keyspace; @@ -56,7 +51,6 @@ import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.utils.FBUtilities; -import org.apache.cassandra.utils.Throwables; /** * TestCluster creates, initializes and manages Cassandra instances ({@link Instance}. @@ -84,7 +78,10 @@ import org.apache.cassandra.utils.Throwables; */ public class TestCluster implements AutoCloseable { - private final ExecutorService exec = Executors.newCachedThreadPool(new NamedThreadFactory("cluster-async-tasks")); + // WARNING: we have this logger not (necessarily) for logging, but + // to ensure we have instantiated the main classloader's LoggerFactory (and any LogbackStatusListener) + // before we instantiate any for a new instance + private static final Logger logger = LoggerFactory.getLogger(TestCluster.class); private final File root; private final List instances; @@ -104,7 +101,7 @@ public class TestCluster implements AutoCloseable void launch() { FBUtilities.waitOnFutures(instances.stream() - .map(i -> exec.submit(() -> i.launch(this))) + .map(i -> i.isolatedExecutor.submit(() -> i.launch(this))) .collect(Collectors.toList()) ); for (Instance instance : instances) @@ -247,37 +244,17 @@ public class TestCluster implements AutoCloseable } @Override - public void close() throws InterruptedException, TimeoutException, ExecutionException + public void close() { List> futures = instances.stream() - .map(i -> exec.submit(i::shutdown)) + .map(i -> i.isolatedExecutor.submit(i::shutdown)) .collect(Collectors.toList()); // Make sure to only delete directory when threads are stopped - Future combined = exec.submit(() -> { - FBUtilities.waitOnFutures(futures); - FileUtils.deleteRecursive(root); - }); - - combined.get(60, TimeUnit.SECONDS); - - exec.shutdownNow(); - exec.awaitTermination(10, TimeUnit.SECONDS); - -// withThreadLeakCheck(futures); - Set threadSet = Thread.getAllStackTraces().keySet(); - for (Thread thread : threadSet) - { - if (thread instanceof FastThreadLocalThread) - ((FastThreadLocalThread)thread).setThreadLocalMap(null); - } - - InternalThreadLocalMap.remove(); - InternalThreadLocalMap.destroy(); - - FastThreadLocal.removeAll(); - FastThreadLocal.destroy(); + FBUtilities.waitOnFutures(futures, 60, TimeUnit.SECONDS); + FileUtils.deleteRecursive(root); + //withThreadLeakCheck(futures); System.gc(); } @@ -299,26 +276,5 @@ public class TestCluster implements AutoCloseable } } - public void runAndWait(Instance.ThrowingRunnable runnable) - { - try - { - exec.submit(() -> { - try - { - runnable.run(); - } - catch (Throwable throwable) - { - Throwables.maybeFail(throwable); - } - }).get(); - } - catch (Exception e) - { - Throwables.maybeFail(e.getCause()); - } - } - }