diff --git a/CHANGES.txt b/CHANGES.txt index 617bcc49cd..772446d373 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -37,6 +37,7 @@ Merged from 2.2: * Refactor Circle CI configuration (CASSANDRA-14806) * Fixing invalid CQL in security documentation (CASSANDRA-15020) * Multi-version in-JVM dtests (CASSANDRA-14937) + * Allow instance class loaders to be garbage collected for inJVM dtest (CASSANDRA-15170) 3.11.4 diff --git a/build.xml b/build.xml index d8f833ac85..0b4e42327a 100644 --- a/build.xml +++ b/build.xml @@ -1852,10 +1852,24 @@ - + - + + + + + + + + + + + + + diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java index c38a4af722..3c0ad564b0 100644 --- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java +++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java @@ -35,6 +35,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ArrayListMultimap; @@ -71,6 +72,7 @@ import org.apache.cassandra.net.MessageOut; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.WriteResponseHandler; +import org.apache.cassandra.utils.ExecutorUtils; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MBeanWrapper; import org.apache.cassandra.utils.UUIDGen; @@ -112,10 +114,9 @@ public class BatchlogManager implements BatchlogManagerMBean TimeUnit.MILLISECONDS); } - public void shutdown() throws InterruptedException + public void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { - batchlogTasks.shutdown(); - batchlogTasks.awaitTermination(60, TimeUnit.SECONDS); + ExecutorUtils.shutdownAndWait(timeout, unit, batchlogTasks); } public static void remove(UUID id) diff --git a/src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java b/src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java index 199803f044..b54fa3fca5 100644 --- a/src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java @@ -70,7 +70,7 @@ public class InfiniteLoopExecutor return this; } - public void shutdown() + public void shutdownNow() { isShutdown = true; thread.interrupt(); diff --git a/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java b/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java index 5e3e5cf396..3da4569a4d 100644 --- a/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java +++ b/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java @@ -19,9 +19,12 @@ package org.apache.cassandra.concurrent; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import com.google.common.annotations.VisibleForTesting; +import org.apache.cassandra.utils.ExecutorUtils; + /** * Centralized location for shared executors */ @@ -48,12 +51,8 @@ public class ScheduledExecutors public static final DebuggableScheduledThreadPoolExecutor optionalTasks = new DebuggableScheduledThreadPoolExecutor("OptionalTasks"); @VisibleForTesting - public static void shutdownAndWait() throws InterruptedException + public static void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { - ExecutorService[] executors = new ExecutorService[] { scheduledFastTasks, scheduledTasks, nonPeriodicTasks, optionalTasks }; - for (ExecutorService executor : executors) - executor.shutdownNow(); - for (ExecutorService executor : executors) - executor.awaitTermination(60, TimeUnit.SECONDS); + ExecutorUtils.shutdownNowAndWait(timeout, unit, scheduledFastTasks, scheduledTasks, nonPeriodicTasks, optionalTasks); } } diff --git a/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java b/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java index 65a2a4907b..f309b46476 100644 --- a/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java +++ b/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java @@ -115,7 +115,7 @@ public class SharedExecutorPool return executor; } - public synchronized void shutdownAndWait() throws InterruptedException + public synchronized void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException { shuttingDown = true; List executors = new ArrayList<>(this.executors); @@ -124,7 +124,7 @@ public class SharedExecutorPool terminateWorkers(); - long until = System.nanoTime() + TimeUnit.MINUTES.toNanos(1L); + long until = System.nanoTime() + unit.toNanos(timeout); for (SEPExecutor executor : executors) executor.shutdown.await(until - System.nanoTime(), TimeUnit.NANOSECONDS); } diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java index 7f59b4bbb6..857c5b7554 100644 --- a/src/java/org/apache/cassandra/concurrent/StageManager.java +++ b/src/java/org/apache/cassandra/concurrent/StageManager.java @@ -25,6 +25,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.utils.ExecutorUtils; import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.config.DatabaseDescriptor.*; @@ -114,12 +115,9 @@ public class StageManager } @VisibleForTesting - public static void shutdownAndWait() throws InterruptedException + public static void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { - for (Stage stage : Stage.values()) - StageManager.stages.get(stage).shutdownNow(); - for (Stage stage : Stage.values()) - StageManager.stages.get(stage).awaitTermination(60, TimeUnit.SECONDS); + ExecutorUtils.shutdownNowAndWait(timeout, unit, StageManager.stages.values()); } /** diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 41b5e73323..89fb2718da 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -83,6 +83,8 @@ import org.apache.cassandra.utils.memory.MemtableAllocator; import org.json.simple.JSONArray; import org.json.simple.JSONObject; +import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination; +import static org.apache.cassandra.utils.ExecutorUtils.shutdown; import static org.apache.cassandra.utils.Throwables.maybeFail; public class ColumnFamilyStore implements ColumnFamilyStoreMBean @@ -264,18 +266,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean postFlushExecutor.awaitTermination(60, TimeUnit.SECONDS); } - public static void shutdownReclaimExecutor() throws InterruptedException + public static void shutdownExecutorsAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { - reclaimExecutor.shutdown(); - reclaimExecutor.awaitTermination(60, TimeUnit.SECONDS); - } - - public static void shutdownPerDiskFlushExecutors() throws InterruptedException - { - for (ExecutorService executorService : perDiskflushExecutors) - executorService.shutdown(); - for (ExecutorService executorService : perDiskflushExecutors) - executorService.awaitTermination(60, TimeUnit.SECONDS); + List executors = new ArrayList<>(perDiskflushExecutors.length + 3); + Collections.addAll(executors, reclaimExecutor, postFlushExecutor, flushExecutor); + Collections.addAll(executors, perDiskflushExecutors); + ExecutorUtils.shutdownAndWait(timeout, unit, executors); } public void reload() diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index 856dafaa57..a9ac968d35 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -390,6 +390,7 @@ public class CommitLog implements CommitLogMBean /** * Shuts down the threads used by the commit log, blocking until completion. + * TODO this should accept a timeout, and throw TimeoutException */ public void shutdownBlocking() throws InterruptedException { diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 6a862e5ed8..a205140f48 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -35,6 +35,7 @@ import com.google.common.util.concurrent.ListenableFutureTask; import com.google.common.util.concurrent.Uninterruptibles; import io.netty.util.concurrent.FastThreadLocal; +import org.apache.cassandra.utils.ExecutorUtils; import org.apache.cassandra.utils.MBeanWrapper; import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.Pair; @@ -56,6 +57,8 @@ import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.CassandraVersion; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.JVMStabilityInspector; +import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination; +import static org.apache.cassandra.utils.ExecutorUtils.shutdown; /** * This module is responsible for Gossiping information for the local endpoint. This abstraction @@ -1808,4 +1811,10 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean logger.info("No gossip backlog; proceeding"); } + @VisibleForTesting + public void stopShutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException + { + stop(); + ExecutorUtils.shutdownAndWait(timeout, unit, executor); + } } diff --git a/src/java/org/apache/cassandra/hints/HintsCatalog.java b/src/java/org/apache/cassandra/hints/HintsCatalog.java index 7d5c8e6006..5a92889ff4 100644 --- a/src/java/org/apache/cassandra/hints/HintsCatalog.java +++ b/src/java/org/apache/cassandra/hints/HintsCatalog.java @@ -20,6 +20,7 @@ package org.apache.cassandra.hints; import java.io.File; import java.io.IOException; import java.nio.file.Files; +import java.nio.file.Path; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Stream; @@ -64,10 +65,10 @@ final class HintsCatalog */ static HintsCatalog load(File hintsDirectory, ImmutableMap writerParams) { - try + try(Stream list = Files.list(hintsDirectory.toPath())) { Map> stores = - Files.list(hintsDirectory.toPath()) + list .filter(HintsDescriptor::isHintFileName) .map(HintsDescriptor::readFromFileQuietly) .filter(Optional::isPresent) diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java index bd52bcaf19..c603404d42 100644 --- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java +++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java @@ -65,6 +65,9 @@ import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.concurrent.OpOrder; import org.apache.cassandra.utils.concurrent.Refs; +import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination; +import static org.apache.cassandra.utils.ExecutorUtils.shutdown; + /** * Handles the core maintenance functionality associated with indexes: adding/removing them to or from * a table, (re)building during bootstrap or other streaming operations, flushing, reloading metadata @@ -1176,12 +1179,10 @@ public class SecondaryIndexManager implements IndexRegistry } @VisibleForTesting - public static void shutdownExecutors() throws InterruptedException + public static void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { ExecutorService[] executors = new ExecutorService[]{ asyncExecutor, blockingExecutor }; - for (ExecutorService executor : executors) - executor.shutdown(); - for (ExecutorService executor : executors) - executor.awaitTermination(60, TimeUnit.SECONDS); + shutdown(executors); + awaitTermination(timeout, unit, executors); } } diff --git a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java index 507b6faac0..ae79217ee1 100644 --- a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java +++ b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java @@ -26,6 +26,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableSet; @@ -44,11 +45,15 @@ import org.apache.cassandra.db.lifecycle.LifecycleTransaction; import org.apache.cassandra.db.lifecycle.SSTableSet; import org.apache.cassandra.db.lifecycle.View; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.utils.ExecutorUtils; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MBeanWrapper; import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.WrappedRunnable; +import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination; +import static org.apache.cassandra.utils.ExecutorUtils.shutdown; + /** * Manages the fixed-size memory pool for index summaries, periodically resizing them * in order to give more memory to hot sstables and less memory to cold sstables. @@ -258,4 +263,10 @@ public class IndexSummaryManager implements IndexSummaryManagerMBean { return CompactionManager.instance.runIndexSummaryRedistribution(redistribution); } + + @VisibleForTesting + public void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException + { + ExecutorUtils.shutdownAndWait(timeout, unit, executor); + } } diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java index 6866ff9075..06f0072137 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java @@ -2371,10 +2371,10 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted executors = ImmutableList.of(keepAliveExecutor); + ExecutorUtils.shutdownNow(executors); + ExecutorUtils.awaitTermination(timeout, unit, executors); + } } diff --git a/src/java/org/apache/cassandra/utils/ExecutorUtils.java b/src/java/org/apache/cassandra/utils/ExecutorUtils.java new file mode 100644 index 0000000000..21933a3123 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/ExecutorUtils.java @@ -0,0 +1,151 @@ +/* + * 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.util.Arrays; +import java.util.Collection; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.apache.cassandra.concurrent.InfiniteLoopExecutor; + +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +public class ExecutorUtils +{ + + public static Runnable runWithThreadName(Runnable runnable, String threadName) + { + return () -> { + String oldThreadName = Thread.currentThread().getName(); + try + { + Thread.currentThread().setName(threadName); + runnable.run(); + } + finally + { + Thread.currentThread().setName(oldThreadName); + } + }; + } + + public static void shutdownNow(Iterable executors) + { + shutdown(true, executors); + } + + public static void shutdown(Iterable executors) + { + shutdown(false, executors); + } + + public static void shutdown(boolean interrupt, Iterable executors) + { + for (Object executor : executors) + { + if (executor instanceof ExecutorService) + { + if (interrupt) ((ExecutorService) executor).shutdownNow(); + else ((ExecutorService) executor).shutdown(); + } + else if (executor instanceof InfiniteLoopExecutor) + ((InfiniteLoopExecutor) executor).shutdownNow(); + else if (executor instanceof Thread) + ((Thread) executor).interrupt(); + else if (executor != null) + throw new IllegalArgumentException(executor.toString()); + } + } + + public static void shutdown(ExecutorService ... executors) + { + shutdown(Arrays.asList(executors)); + } + + public static void shutdownNow(ExecutorService ... executors) + { + shutdownNow(Arrays.asList(executors)); + } + + public static void awaitTermination(long timeout, TimeUnit unit, ExecutorService ... executors) throws InterruptedException, TimeoutException + { + awaitTermination(timeout, unit, Arrays.asList(executors)); + } + + public static void awaitTermination(long timeout, TimeUnit unit, Collection executors) throws InterruptedException, TimeoutException + { + long deadline = System.nanoTime() + unit.toNanos(timeout); + awaitTerminationUntil(deadline, executors); + } + + public static void awaitTerminationUntil(long deadline, Collection executors) throws InterruptedException, TimeoutException + { + for (Object executor : executors) + { + long wait = deadline - System.nanoTime(); + if (executor instanceof ExecutorService) + { + if (wait <= 0 || !((ExecutorService)executor).awaitTermination(wait, NANOSECONDS)) + throw new TimeoutException(executor + " did not terminate on time"); + } + else if (executor instanceof InfiniteLoopExecutor) + { + if (wait <= 0 || !((InfiniteLoopExecutor)executor).awaitTermination(wait, NANOSECONDS)) + throw new TimeoutException(executor + " did not terminate on time"); + } + else if (executor instanceof Thread) + { + Thread t = (Thread) executor; + if (wait <= 0) + throw new TimeoutException(executor + " did not terminate on time"); + t.join((wait + 999999) / 1000000L, (int) (wait % 1000000L)); + if (t.isAlive()) + throw new TimeoutException(executor + " did not terminate on time"); + } + else if (executor != null) + { + throw new IllegalArgumentException(executor.toString()); + } + } + } + + public static void shutdownAndWait(long timeout, TimeUnit unit, Collection executors) throws TimeoutException, InterruptedException + { + shutdown(executors); + awaitTermination(timeout, unit, executors); + } + + public static void shutdownNowAndWait(long timeout, TimeUnit unit, Collection executors) throws TimeoutException, InterruptedException + { + shutdownNow(executors); + awaitTermination(timeout, unit, executors); + } + + public static void shutdownAndWait(long timeout, TimeUnit unit, Object ... executors) throws TimeoutException, InterruptedException + { + shutdownAndWait(timeout, unit, Arrays.asList(executors)); + } + + public static void shutdownNowAndWait(long timeout, TimeUnit unit, Object ... executors) throws TimeoutException, InterruptedException + { + shutdownNowAndWait(timeout, unit, Arrays.asList(executors)); + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/utils/concurrent/Ref.java b/src/java/org/apache/cassandra/utils/concurrent/Ref.java index 84888d1cd6..5b6c3d6056 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Ref.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Ref.java @@ -45,12 +45,16 @@ import org.apache.cassandra.db.lifecycle.View; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.Memory; import org.apache.cassandra.io.util.SafeMemory; +import org.apache.cassandra.utils.ExecutorUtils; import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.Pair; import org.cliffc.high_scale_lib.NonBlockingHashMap; import static java.util.Collections.emptyList; +import org.apache.cassandra.concurrent.InfiniteLoopExecutor.InterruptibleRunnable; +import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination; +import static org.apache.cassandra.utils.ExecutorUtils.shutdownNow; import static org.apache.cassandra.utils.Throwables.maybeFail; import static org.apache.cassandra.utils.Throwables.merge; @@ -705,14 +709,8 @@ public final class Ref implements RefCounted } @VisibleForTesting - public static void shutdownReferenceReaper() throws InterruptedException + public static void shutdownReferenceReaper(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { - EXEC.shutdown(); - EXEC.awaitTermination(60, TimeUnit.SECONDS); - if (STRONG_LEAK_DETECTOR != null) - { - STRONG_LEAK_DETECTOR.shutdownNow(); - STRONG_LEAK_DETECTOR.awaitTermination(60, TimeUnit.SECONDS); - } + ExecutorUtils.shutdownNowAndWait(timeout, unit, EXEC, STRONG_LEAK_DETECTOR); } } diff --git a/src/java/org/apache/cassandra/utils/memory/BufferPool.java b/src/java/org/apache/cassandra/utils/memory/BufferPool.java index c8ad078527..e91c9e2a52 100644 --- a/src/java/org/apache/cassandra/utils/memory/BufferPool.java +++ b/src/java/org/apache/cassandra/utils/memory/BufferPool.java @@ -21,6 +21,7 @@ package org.apache.cassandra.utils.memory; import java.lang.ref.PhantomReference; import java.lang.ref.ReferenceQueue; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Queue; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicLong; @@ -40,6 +41,9 @@ import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.concurrent.Ref; +import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination; +import static org.apache.cassandra.utils.ExecutorUtils.shutdownNow; + /** * A pool of ByteBuffers that can be recycled. */ @@ -856,9 +860,9 @@ public class BufferPool } @VisibleForTesting - public static void shutdownLocalCleaner() throws InterruptedException + public static void shutdownLocalCleaner(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { - EXEC.shutdown(); - EXEC.awaitTermination(60, TimeUnit.SECONDS); + shutdownNow(Arrays.asList(EXEC)); + awaitTermination(timeout, unit, Arrays.asList(EXEC)); } } diff --git a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java index 50b6f106c9..bd17f78124 100644 --- a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java +++ b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java @@ -19,6 +19,7 @@ package org.apache.cassandra.utils.memory; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicLongFieldUpdater; import com.google.common.annotations.VisibleForTesting; @@ -27,6 +28,7 @@ import com.codahale.metrics.Timer; import org.apache.cassandra.metrics.CassandraMetricsRegistry; import org.apache.cassandra.metrics.DefaultNameFactory; import org.apache.cassandra.utils.concurrent.WaitQueue; +import org.apache.cassandra.utils.ExecutorUtils; /** @@ -67,12 +69,12 @@ public abstract class MemtablePool } @VisibleForTesting - public void shutdown() throws InterruptedException + public void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { - cleaner.shutdown(); - cleaner.awaitTermination(60, TimeUnit.SECONDS); + ExecutorUtils.shutdownNowAndWait(timeout, unit, cleaner); } + public abstract MemtableAllocator newAllocator(); /** diff --git a/test/conf/logback-dtest.xml b/test/conf/logback-dtest.xml index b62539fb3c..4282feec4b 100644 --- a/test/conf/logback-dtest.xml +++ b/test/conf/logback-dtest.xml @@ -23,7 +23,7 @@ - + ./build/test/logs/${cassandra.testtag}/TEST-${suitename}.log @@ -42,14 +42,14 @@ false - + 0 0 1024 - + - + %-5level %date{HH:mm:ss,SSS} %msg%n @@ -58,7 +58,7 @@ - + %-5level %date{HH:mm:ss,SSS} %msg%n @@ -67,7 +67,7 @@ - + %-5level [%thread] ${instance_id} %date{ISO8601} %F:%L - %msg%n @@ -79,8 +79,8 @@ - - - + + + diff --git a/test/distributed/org/apache/cassandra/distributed/Cluster.java b/test/distributed/org/apache/cassandra/distributed/Cluster.java index 4ae4e5d340..95862b6a62 100644 --- a/test/distributed/org/apache/cassandra/distributed/Cluster.java +++ b/test/distributed/org/apache/cassandra/distributed/Cluster.java @@ -19,10 +19,10 @@ package org.apache.cassandra.distributed; import java.io.File; +import java.io.IOException; import java.util.List; -import java.util.Set; +import java.util.function.Consumer; -import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.ICluster; import org.apache.cassandra.distributed.impl.AbstractCluster; import org.apache.cassandra.distributed.impl.IInvokableInstance; @@ -35,31 +35,29 @@ import org.apache.cassandra.distributed.impl.Versions; */ public class Cluster extends AbstractCluster implements ICluster, AutoCloseable { - private Cluster(File root, Versions.Version version, List configs, Set features, ClassLoader sharedClassLoader) + private Cluster(File root, Versions.Version version, List configs, ClassLoader sharedClassLoader) { - super(root, version, configs, features, sharedClassLoader); + super(root, version, configs, sharedClassLoader); } - protected IInvokableInstance newInstanceWrapper(Versions.Version version, InstanceConfig config) + protected IInvokableInstance newInstanceWrapper(int generation, Versions.Version version, InstanceConfig config) { - return new Wrapper(version, config); + return new Wrapper(generation, version, config); + } + + public static Builder build(int nodeCount) + { + return new Builder<>(nodeCount, Cluster::new); + } + + public static Cluster create(int nodeCount, Consumer configUpdater) throws IOException + { + return build(nodeCount).withConfig(configUpdater).start(); } public static Cluster create(int nodeCount) throws Throwable { - return create(nodeCount, Cluster::new); - } - public static Cluster create(int nodeCount, Set with) throws Throwable - { - return create(nodeCount, with, Cluster::new); - } - public static Cluster create(int nodeCount, File root) - { - return create(nodeCount, root, Cluster::new); - } - public static Cluster create(int nodeCount, File root, Set with) - { - return create(nodeCount, Versions.CURRENT, root, with, Cluster::new); + return build(nodeCount).start(); } } diff --git a/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java b/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java index d0613b10a9..c77d72514b 100644 --- a/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java @@ -19,12 +19,8 @@ package org.apache.cassandra.distributed; import java.io.File; -import java.io.IOException; -import java.nio.file.Files; import java.util.List; -import java.util.Set; -import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.ICluster; import org.apache.cassandra.distributed.impl.AbstractCluster; import org.apache.cassandra.distributed.impl.IUpgradeableInstance; @@ -40,35 +36,29 @@ import org.apache.cassandra.distributed.impl.Versions; */ public class UpgradeableCluster extends AbstractCluster implements ICluster, AutoCloseable { - private UpgradeableCluster(File root, Versions.Version version, List configs, Set features, ClassLoader sharedClassLoader) + private UpgradeableCluster(File root, Versions.Version version, List configs, ClassLoader sharedClassLoader) { - super(root, version, configs, features, sharedClassLoader); + super(root, version, configs, sharedClassLoader); } - protected IUpgradeableInstance newInstanceWrapper(Versions.Version version, InstanceConfig config) + protected IUpgradeableInstance newInstanceWrapper(int generation, Versions.Version version, InstanceConfig config) { - return new Wrapper(version, config); + return new Wrapper(generation, version, config); + } + + public static Builder build(int nodeCount) + { + return new Builder<>(nodeCount, UpgradeableCluster::new); } public static UpgradeableCluster create(int nodeCount) throws Throwable { - return create(nodeCount, UpgradeableCluster::new); + return build(nodeCount).start(); } - public static UpgradeableCluster create(int nodeCount, File root) + public static UpgradeableCluster create(int nodeCount, Versions.Version version) throws Throwable { - return create(nodeCount, root, UpgradeableCluster::new); + return build(nodeCount).withVersion(version).start(); } - - public static UpgradeableCluster create(int nodeCount, Versions.Version version) throws IOException - { - return create(nodeCount, version, UpgradeableCluster::new); - } - - public static UpgradeableCluster create(int nodeCount, Versions.Version version, File root) - { - return create(nodeCount, version, root, UpgradeableCluster::new); - } - } diff --git a/test/distributed/org/apache/cassandra/distributed/api/IInstance.java b/test/distributed/org/apache/cassandra/distributed/api/IInstance.java index 25e2c94e4d..d5382b4970 100644 --- a/test/distributed/org/apache/cassandra/distributed/api/IInstance.java +++ b/test/distributed/org/apache/cassandra/distributed/api/IInstance.java @@ -20,7 +20,6 @@ package org.apache.cassandra.distributed.api; import org.apache.cassandra.locator.InetAddressAndPort; -import java.util.Set; import java.util.UUID; import java.util.concurrent.Future; @@ -43,7 +42,7 @@ public interface IInstance extends IIsolatedExecutor Future shutdown(boolean graceful); // these methods are not for external use, but for simplicity we leave them public and on the normal IInstance interface - void startup(ICluster cluster, Set with); + void startup(ICluster cluster); void receiveMessage(IMessage message); int getMessagingVersion(); diff --git a/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java index 6741b3fdd5..3e5a18fe7c 100644 --- a/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java +++ b/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java @@ -38,4 +38,5 @@ public interface IInstanceConfig Object get(String fieldName); String getString(String fieldName); int getInt(String fieldName); + boolean has(Feature featureFlag); } diff --git a/test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java b/test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java index b5fde840e7..f7c80940a8 100644 --- a/test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java +++ b/test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java @@ -18,11 +18,8 @@ package org.apache.cassandra.distributed.api; -import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.net.MessagingService; -import java.util.function.BiConsumer; - public interface IMessageFilters { public interface Filter @@ -44,5 +41,6 @@ public interface IMessageFilters void reset(); // internal - BiConsumer filter(BiConsumer applyIfNotFiltered); + boolean permit(IInstance from, IInstance to, int verb); + } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index 67c844f7a2..d09ed9d819 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -32,6 +32,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -44,7 +45,6 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.Keyspace; -import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.ICoordinator; import org.apache.cassandra.distributed.api.IInstance; import org.apache.cassandra.distributed.api.IInstanceConfig; @@ -89,10 +89,10 @@ public abstract class AbstractCluster implements ICluster, // to ensure we have instantiated the main classloader's LoggerFactory (and any LogbackStatusListener) // before we instantiate any for a new instance private static final Logger logger = LoggerFactory.getLogger(AbstractCluster.class); + private static final AtomicInteger generation = new AtomicInteger(); private final File root; private final ClassLoader sharedClassLoader; - private final Set features; // mutated by starting/stopping a node private final List instances; @@ -103,6 +103,7 @@ public abstract class AbstractCluster implements ICluster, protected class Wrapper extends DelegatingInvokableInstance implements IUpgradeableInstance { + private final int generation; private final InstanceConfig config; private volatile IInvokableInstance delegate; private volatile Versions.Version version; @@ -111,21 +112,22 @@ public abstract class AbstractCluster implements ICluster, protected IInvokableInstance delegate() { if (delegate == null) - delegate = newInstance(); + delegate = newInstance(generation); return delegate; } - public Wrapper(Versions.Version version, InstanceConfig config) + public Wrapper(int generation, Versions.Version version, InstanceConfig config) { + this.generation = generation; this.config = config; this.version = version; // we ensure there is always a non-null delegate, so that the executor may be used while the node is offline - this.delegate = newInstance(); + this.delegate = newInstance(generation); } - private IInvokableInstance newInstance() + private IInvokableInstance newInstance(int generation) { - ClassLoader classLoader = new InstanceClassLoader(config.num(), version.classpath, sharedClassLoader); + ClassLoader classLoader = new InstanceClassLoader(generation, version.classpath, sharedClassLoader); return Instance.transferAdhoc((SerializableBiFunction)Instance::new, classLoader) .apply(config, classLoader); } @@ -145,11 +147,17 @@ public abstract class AbstractCluster implements ICluster, { if (!isShutdown) throw new IllegalStateException(); - delegate().startup(AbstractCluster.this, features); + delegate().startup(AbstractCluster.this); isShutdown = false; updateMessagingVersions(); } + @Override + public synchronized Future shutdown() + { + return shutdown(true); + } + @Override public synchronized Future shutdown(boolean graceful) { @@ -185,26 +193,26 @@ public abstract class AbstractCluster implements ICluster, } } - protected AbstractCluster(File root, Versions.Version version, List configs, Set features, ClassLoader sharedClassLoader) + protected AbstractCluster(File root, Versions.Version version, List configs, ClassLoader sharedClassLoader) { this.root = root; - this.features = features; this.sharedClassLoader = sharedClassLoader; this.instances = new ArrayList<>(); this.instanceMap = new HashMap<>(); + int generation = AbstractCluster.generation.incrementAndGet(); for (InstanceConfig config : configs) { - I instance = newInstanceWrapper(version, config); + I instance = newInstanceWrapper(generation, version, config); instances.add(instance); // we use the config().broadcastAddressAndPort() here because we have not initialised the Instance I prev = instanceMap.put(instance.broadcastAddressAndPort(), instance); if (null != prev) throw new IllegalStateException("Cluster cannot have multiple nodes with same InetAddressAndPort: " + instance.broadcastAddressAndPort() + " vs " + prev.broadcastAddressAndPort()); } - this.filters = new MessageFilters(this); + this.filters = new MessageFilters(); } - protected abstract I newInstanceWrapper(Versions.Version version, InstanceConfig config); + protected abstract I newInstanceWrapper(int generation, Versions.Version version, InstanceConfig config); /** * WARNING: we index from 1 here, for consistency with inet address! @@ -262,9 +270,12 @@ public abstract class AbstractCluster implements ICluster, { for (IInstance reportTo: instances) { + if (reportTo.isShutdown()) + continue; + for (IInstance reportFrom: instances) { - if (reportFrom == reportTo) + if (reportFrom == reportTo || reportFrom.isShutdown()) continue; int minVersion = Math.min(reportFrom.getMessagingVersion(), reportTo.getMessagingVersion()); @@ -337,71 +348,86 @@ public abstract class AbstractCluster implements ICluster, protected interface Factory> { - C newCluster(File root, Versions.Version version, List configs, Set features, ClassLoader sharedClassLoader); + C newCluster(File root, Versions.Version version, List configs, ClassLoader sharedClassLoader); } - protected static > C - create(int nodeCount, Factory factory) throws Throwable + public static class Builder> { - return create(nodeCount, Collections.emptySet(), factory); - } - - protected static > C - create(int nodeCount, Set features, Factory factory) throws Throwable - { - return create(nodeCount, Files.createTempDirectory("dtests").toFile(), features, factory); - } - - protected static > C - create(int nodeCount, File root, Factory factory) - { - return create(nodeCount, root, Collections.emptySet(), factory); - } - - protected static > C - create(int nodeCount, File root, Set features, Factory factory) - { - return create(nodeCount, Versions.CURRENT, root, features, factory); - } - - protected static > C - create(int nodeCount, Versions.Version version, Factory factory) throws IOException - { - return create(nodeCount, version, Collections.emptySet(), factory); - } - - protected static > C - create(int nodeCount, Versions.Version version, Set features, Factory factory) throws IOException - { - return create(nodeCount, version, Files.createTempDirectory("dtests").toFile(), features, factory); - } - - protected static > C - create(int nodeCount, Versions.Version version, File root, Factory factory) - { - return create(nodeCount, version, root, Collections.emptySet(), factory); - } - protected static > C - create(int nodeCount, Versions.Version version, File root, Set features, Factory factory) - { - root.mkdirs(); - setupLogging(root); - - ClassLoader sharedClassLoader = Thread.currentThread().getContextClassLoader(); - - List configs = new ArrayList<>(); - long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount); - for (int i = 0 ; i < nodeCount ; ++i) + private final int nodeCount; + private final Factory factory; + private int subnet; + private File root; + private Versions.Version version; + private Consumer configUpdater; + public Builder(int nodeCount, Factory factory) { - InstanceConfig config = InstanceConfig.generate(i + 1, root, String.valueOf(token)); - configs.add(config); - token += increment; + this.nodeCount = nodeCount; + this.factory = factory; } - C cluster = factory.newCluster(root, version, configs, features, sharedClassLoader); - return cluster; + public Builder withSubnet(int subnet) + { + this.subnet = subnet; + return this; + } + + public Builder withRoot(File root) + { + this.root = root; + return this; + } + + public Builder withVersion(Versions.Version version) + { + this.version = version; + return this; + } + + public Builder withConfig(Consumer updater) + { + this.configUpdater = updater; + return this; + } + + public C createWithoutStarting() throws IOException + { + File root = this.root; + Versions.Version version = this.version; + + if (root == null) + root = Files.createTempDirectory("dtests").toFile(); + if (version == null) + version = Versions.CURRENT; + + root.mkdirs(); + setupLogging(root); + + ClassLoader sharedClassLoader = Thread.currentThread().getContextClassLoader(); + + List configs = new ArrayList<>(); + long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount); + for (int i = 0; i < nodeCount; ++i) + { + InstanceConfig config = InstanceConfig.generate(i + 1, subnet, root, String.valueOf(token)); + if (configUpdater != null) + configUpdater.accept(config); + configs.add(config); + token += increment; + } + + C cluster = factory.newCluster(root, version, configs, sharedClassLoader); + return cluster; + } + + public C start() throws IOException + { + C cluster = createWithoutStarting(); + cluster.startup(); + return cluster; + } } + private static void setupLogging(File root) { try diff --git a/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java b/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java index 94df6cd11d..a1f954b48f 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java @@ -27,7 +27,6 @@ import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; -import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.ICluster; import org.apache.cassandra.distributed.api.ICoordinator; import org.apache.cassandra.distributed.api.IInstanceConfig; @@ -112,9 +111,9 @@ public abstract class DelegatingInvokableInstance implements IInvokableInstance } @Override - public void startup(ICluster cluster, Set with) + public void startup(ICluster cluster) { - delegate().startup(cluster, with); + delegate().startup(cluster); } @Override diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index f209d53000..784eb3fe6e 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -34,9 +34,6 @@ import java.util.concurrent.Future; import java.util.function.BiConsumer; import java.util.function.Function; -import org.slf4j.LoggerFactory; - -import ch.qos.logback.classic.LoggerContext; import org.apache.cassandra.batchlog.BatchlogManager; import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.concurrent.SharedExecutorPool; @@ -55,11 +52,12 @@ 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.db.monitoring.ApproximateTime; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Token; -import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.ICluster; import org.apache.cassandra.distributed.api.ICoordinator; +import org.apache.cassandra.distributed.api.IInstance; import org.apache.cassandra.distributed.api.IInstanceConfig; import org.apache.cassandra.distributed.api.IListen; import org.apache.cassandra.distributed.api.IMessage; @@ -68,6 +66,7 @@ import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.gms.VersionedValue; import org.apache.cassandra.hints.HintsService; import org.apache.cassandra.index.SecondaryIndexManager; +import org.apache.cassandra.io.sstable.IndexSummaryManager; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.DataInputBuffer; import org.apache.cassandra.io.util.DataOutputBuffer; @@ -82,12 +81,19 @@ 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.streaming.StreamCoordinator; +import org.apache.cassandra.streaming.StreamSession; import org.apache.cassandra.transport.messages.ResultMessage; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.NanoTimeToCurrentTimeMillis; import org.apache.cassandra.utils.Throwables; import org.apache.cassandra.utils.concurrent.Ref; import org.apache.cassandra.utils.memory.BufferPool; +import static java.util.concurrent.TimeUnit.MINUTES; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + public class Instance extends IsolatedExecutor implements IInvokableInstance { public final IInstanceConfig config; @@ -100,10 +106,10 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance this.config = config; InstanceIDDefiner.setInstanceId(config.num()); FBUtilities.setBroadcastInetAddress(config.broadcastAddressAndPort().address); - acceptsOnInstance((IInstanceConfig override) -> { - Config.setOverrideLoadConfig(() -> loadConfig(override)); - DatabaseDescriptor.daemonInitialization(); - }).accept(config); + // Set the config at instance creation, possibly before startup() has run on all other instances. + // setMessagingVersions below will call runOnInstance which will instantiate + // the MessagingService and dependencies preventing later changes to network parameters. + Config.setOverrideLoadConfig(() -> loadConfig(config)); } public IInstanceConfig config() @@ -182,7 +188,10 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance private void registerMockMessaging(ICluster cluster) { BiConsumer deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message); - BiConsumer deliverToInstanceIfNotFiltered = cluster.filters().filter(deliverToInstance); + BiConsumer deliverToInstanceIfNotFiltered = (to, message) -> { + if (cluster.filters().permit(this, cluster.get(to), message.verb())) + deliverToInstance.accept(to, message); + }; Map addressAndPortMap = new HashMap<>(); cluster.stream().forEach(instance -> { @@ -198,6 +207,26 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance new MessageDeliverySink(deliverToInstanceIfNotFiltered, addressAndPortMap::get)); } + // unnecessary if registerMockMessaging used + private void registerFilter(ICluster cluster) + { + IInstance instance = this; + MessagingService.instance().addMessageSink(new IMessageSink() + { + public boolean allowOutgoingMessage(MessageOut message, int id, InetAddress toAddress) + { + // Port is not passed in, so take a best guess at the destination port from this instance + IInstance to = cluster.get(InetAddressAndPort.getByAddressOverrideDefaults(toAddress, instance.config().broadcastAddressAndPort().port)); + return cluster.filters().permit(instance, to, message.verb.ordinal()); + } + + public boolean allowIncomingMessage(MessageIn message, int id) + { + return true; + } + }); + } + private class MessageDeliverySink implements IMessageSink { private final BiConsumer deliver; @@ -216,6 +245,11 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance assert from.equals(lookupAddressAndPort.apply(messageOut.from)); InetAddressAndPort toFull = lookupAddressAndPort.apply(to); int version = MessagingService.instance().getVersion(to); + + out.writeInt(MessagingService.PROTOCOL_MAGIC); + out.writeInt(id); + long timestamp = System.currentTimeMillis(); + out.writeInt((int) timestamp); messageOut.serialize(out, version); deliver.accept(toFull, new Message(messageOut.verb.ordinal(), out.toByteArray(), id, version, from)); } @@ -233,14 +267,32 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance } } - public void receiveMessage(IMessage message) + public void receiveMessage(IMessage imessage) { sync(() -> { - try (DataInputBuffer in = new DataInputBuffer(message.bytes())) + // Based on org.apache.cassandra.net.IncomingTcpConnection.receiveMessage + try (DataInputBuffer input = new DataInputBuffer(imessage.bytes())) { - MessageIn messageIn = MessageIn.read(in, message.version(), message.id()); - Runnable deliver = new MessageDeliveryTask(messageIn, message.id()); - deliver.run(); + int version = imessage.version(); + + MessagingService.validateMagic(input.readInt()); + int id; + if (version < MessagingService.VERSION_20) + id = Integer.parseInt(input.readUTF()); + else + id = input.readInt(); + long currentTime = ApproximateTime.currentTimeMillis(); + MessageIn message = MessageIn.read(input, version, id, MessageIn.readConstructionTime(imessage.from().address, input, currentTime)); + if (message == null) + { + // callback expired; nothing to do + return; + } + if (version <= MessagingService.current_version) + { + MessagingService.instance().receive(message, id); + } + // else ignore message } catch (Throwable t) { @@ -260,12 +312,14 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance } @Override - public void startup(ICluster cluster, Set with) + public void startup(ICluster cluster) { sync(() -> { try { mkdirs(); + + DatabaseDescriptor.daemonInitialization(); DatabaseDescriptor.createAllDirectories(); // We need to persist this as soon as possible after startup checks. @@ -294,17 +348,27 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance { throw new RuntimeException(e); } - - // TODO: support each separately - if (with.contains(Feature.GOSSIP) || with.contains(Feature.NETWORK)) + if (config.has(NETWORK)) { - StorageService.instance.prepareToJoin(); - StorageService.instance.joinTokenRing(5000); + registerFilter(cluster); + MessagingService.instance().listen(); + } + else + { + // Even though we don't use MessagingService, access the static SocketFactory + // instance here so that we start the static event loop state +// -- not sure what that means? SocketFactory.instance.getClass(); + registerMockMessaging(cluster); + } + + // TODO: this is more than just gossip + if (config.has(GOSSIP)) + { + StorageService.instance.initServer(); } else { initializeRing(cluster); - registerMockMessaging(cluster); } SystemKeyspace.finishStartup(); @@ -404,36 +468,40 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance Future future = async((ExecutorService executor) -> { Throwable error = null; + + if (config.has(GOSSIP) || config.has(NETWORK)) + { + StorageService.instance.shutdownServer(); + } + error = parallelRun(error, executor, - Gossiper.instance::stop, - CompactionManager.instance::forceShutdown, - BatchlogManager.instance::shutdown, - HintsService.instance::shutdownBlocking, - SecondaryIndexManager::shutdownExecutors, - ColumnFamilyStore::shutdownFlushExecutor, - ColumnFamilyStore::shutdownPostFlushExecutor, - ColumnFamilyStore::shutdownReclaimExecutor, - ColumnFamilyStore::shutdownPerDiskFlushExecutors, - PendingRangeCalculatorService.instance::shutdownExecutor, - BufferPool::shutdownLocalCleaner, - Ref::shutdownReferenceReaper, - Memtable.MEMORY_POOL::shutdown, - ScheduledExecutors::shutdownAndWait, - SSTableReader::shutdownBlocking + () -> Gossiper.instance.stopShutdownAndWait(1L, MINUTES), + CompactionManager.instance::forceShutdown, + () -> BatchlogManager.instance.shutdownAndWait(1L, MINUTES), + HintsService.instance::shutdownBlocking, + () -> StreamCoordinator.shutdownAndWait(1L, MINUTES), + () -> StreamSession.shutdownAndWait(1L, MINUTES), + () -> SecondaryIndexManager.shutdownAndWait(1L, MINUTES), + () -> IndexSummaryManager.instance.shutdownAndWait(1L, MINUTES), + () -> ColumnFamilyStore.shutdownExecutorsAndWait(1L, MINUTES), + () -> PendingRangeCalculatorService.instance.shutdownExecutor(1L, MINUTES), + () -> BufferPool.shutdownLocalCleaner(1L, MINUTES), + () -> Ref.shutdownReferenceReaper(1L, MINUTES), + () -> Memtable.MEMORY_POOL.shutdownAndWait(1L, MINUTES), + () -> SSTableReader.shutdownBlocking(1L, MINUTES) ); error = parallelRun(error, executor, + () -> ScheduledExecutors.shutdownAndWait(1L, MINUTES), MessagingService.instance()::shutdown ); error = parallelRun(error, executor, - StageManager::shutdownAndWait, - SharedExecutorPool.SHARED::shutdownAndWait + () -> StageManager.shutdownAndWait(1L, MINUTES), + () -> SharedExecutorPool.SHARED.shutdownAndWait(1L, MINUTES) ); error = parallelRun(error, executor, CommitLog.instance::shutdownBlocking ); - LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); - loggerContext.stop(); Throwables.maybeFail(error); }).apply(isolatedExecutor); diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java index 57530e0934..0ef5a69113 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java @@ -46,6 +46,7 @@ public class InstanceClassLoader extends URLClassLoader name.startsWith("org.apache.cassandra.distributed.api.") || name.startsWith("sun.") || name.startsWith("oracle.") + || name.startsWith("com.intellij.") || name.startsWith("com.sun.") || name.startsWith("com.sun.") || name.startsWith("java.") @@ -61,16 +62,16 @@ public class InstanceClassLoader extends URLClassLoader InstanceClassLoader create(int id, URL[] urls, ClassLoader sharedClassLoader); } - private final int id; private final URL[] urls; + private final int generation; // used to help debug class loader leaks, by helping determine which classloaders should have been collected private final ClassLoader sharedClassLoader; - InstanceClassLoader(int id, URL[] urls, ClassLoader sharedClassLoader) + InstanceClassLoader(int generation, URL[] urls, ClassLoader sharedClassLoader) { super(urls, null); - this.id = id; this.urls = urls; this.sharedClassLoader = sharedClassLoader; + this.generation = generation; } @Override @@ -107,7 +108,7 @@ public class InstanceClassLoader extends URLClassLoader public String toString() { return "InstanceClassLoader{" + - "id=" + id + + "generation=" + generation + ", urls=" + Arrays.toString(urls) + '}'; } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java index 154ffade9b..a21506219f 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java @@ -20,6 +20,7 @@ package org.apache.cassandra.distributed.impl; import org.apache.cassandra.config.Config; import org.apache.cassandra.config.ParameterizedClass; +import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.IInstanceConfig; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.SimpleSeedProvider; @@ -30,6 +31,7 @@ import java.lang.reflect.Field; import java.net.UnknownHostException; import java.util.Arrays; import java.util.Collections; +import java.util.EnumSet; import java.util.Map; import java.util.TreeMap; import java.util.UUID; @@ -45,6 +47,8 @@ public class InstanceConfig implements IInstanceConfig public UUID hostId() { return hostId; } private final Map params = new TreeMap<>(); + private EnumSet featureFlags; + private volatile InetAddressAndPort broadcastAddressAndPort; @Override @@ -103,6 +107,7 @@ public class InstanceConfig implements IInstanceConfig Collections.singletonMap("seeds", "127.0.0.1"))) // legacy parameters .forceSet("commitlog_sync_batch_window_in_ms", 1.0); + this.featureFlags = EnumSet.noneOf(Feature.class); } private InstanceConfig(InstanceConfig copy) @@ -110,6 +115,18 @@ public class InstanceConfig implements IInstanceConfig this.num = copy.num; this.params.putAll(copy.params); this.hostId = copy.hostId; + this.featureFlags = copy.featureFlags; + } + + public InstanceConfig with(Feature featureFlag) + { + featureFlags.add(featureFlag); + return this; + } + + public boolean has(Feature featureFlag) + { + return featureFlags.contains(featureFlag); } public InstanceConfig set(String fieldName, Object value) @@ -200,13 +217,14 @@ public class InstanceConfig implements IInstanceConfig return (String)params.get(name); } - public static InstanceConfig generate(int nodeNum, File root, String token) + public static InstanceConfig generate(int nodeNum, int subnet, File root, String token) { + String ipPrefix = "127.0." + subnet + "."; return new InstanceConfig(nodeNum, - "127.0.0." + nodeNum, - "127.0.0." + nodeNum, - "127.0.0." + nodeNum, - "127.0.0." + nodeNum, + ipPrefix + nodeNum, + ipPrefix + nodeNum, + ipPrefix + nodeNum, + ipPrefix + 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), diff --git a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java index d82c9e49ab..1d26c5dec3 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java @@ -27,28 +27,36 @@ import java.io.Serializable; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.net.URLClassLoader; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; +import org.slf4j.LoggerFactory; + +import ch.qos.logback.classic.LoggerContext; import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.distributed.api.IIsolatedExecutor; +import org.apache.cassandra.utils.ExecutorUtils; public class IsolatedExecutor implements IIsolatedExecutor { final ExecutorService isolatedExecutor; + private final String name; private final ClassLoader classLoader; private final Method deserializeOnInstance; IsolatedExecutor(String name, ClassLoader classLoader) { + this.name = name; this.isolatedExecutor = Executors.newCachedThreadPool(new NamedThreadFactory("isolatedExecutor", Thread.NORM_PRIORITY, classLoader, new ThreadGroup(name))); this.classLoader = classLoader; this.deserializeOnInstance = lookupDeserializeOneObject(classLoader); @@ -57,9 +65,40 @@ public class IsolatedExecutor implements IIsolatedExecutor public Future shutdown() { isolatedExecutor.shutdown(); - ThrowingRunnable.toRunnable(((URLClassLoader) classLoader)::close).run(); - return CompletableFuture.runAsync(ThrowingRunnable.toRunnable(() -> isolatedExecutor.awaitTermination(60, TimeUnit.SECONDS)), - Executors.newSingleThreadExecutor()); + + /* Use a thread pool with a core pool size of zero to terminate the thread as soon as possible + ** so the instance class loader can be garbage collected. Uses a custom thread factory + ** rather than NamedThreadFactory to avoid calling FastThreadLocal.removeAll() in 3.0 and up + ** as it was observed crashing during test failures and made it harder to find the real cause. + */ + ThreadFactory threadFactory = (Runnable r) -> { + Thread t = new Thread(r, name + "_shutdown"); + t.setDaemon(true); + return t; + }; + ExecutorService shutdownExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 0, TimeUnit.SECONDS, + new LinkedBlockingQueue(), threadFactory); + return shutdownExecutor.submit(() -> { + try + { + ExecutorUtils.awaitTermination(60, TimeUnit.SECONDS, isolatedExecutor); + + // Shutdown logging last - this is not ideal as the logging subsystem is initialized + // outsize of this class, however doing it this way provides access to the full + // logging system while termination is taking place. + LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); + loggerContext.stop(); + + // Close the instance class loader after shutting down the isolatedExecutor and logging + // in case error handling triggers loading additional classes + ((URLClassLoader) classLoader).close(); + } + finally + { + shutdownExecutor.shutdownNow(); + } + return null; + }); } public CallableNoExcept> async(CallableNoExcept call) { return () -> isolatedExecutor.submit(call); } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/MessageFilters.java b/test/distributed/org/apache/cassandra/distributed/impl/MessageFilters.java index a72c7a56b4..c1607f8ba0 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/MessageFilters.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/MessageFilters.java @@ -32,33 +32,20 @@ import org.apache.cassandra.net.MessagingService; public class MessageFilters implements IMessageFilters { - private final ICluster cluster; private final Set filters = new CopyOnWriteArraySet<>(); - public MessageFilters(AbstractCluster cluster) + public boolean permit(IInstance from, IInstance to, int verb) { - this.cluster = cluster; - } + if (from == null || to == null) + return false; // cannot deliver + int fromNum = from.config().num(); + int toNum = to.config().num(); - public BiConsumer filter(BiConsumer applyIfNotFiltered) - { - return (toAddress, message) -> - { - IInstance from = cluster.get(message.from()); - IInstance to = cluster.get(toAddress); - if (from == null || to == null) - return; // cannot deliver - int fromNum = from.config().num(); - int toNum = to.config().num(); - int verb = message.verb(); - for (Filter filter : filters) - { - if (filter.matches(fromNum, toNum, verb)) - return; - } + for (Filter filter : filters) + if (filter.matches(fromNum, toNum, verb)) + return false; - applyIfNotFiltered.accept(toAddress, message); - }; + return true; } public class Filter implements IMessageFilters.Filter diff --git a/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java index 3945ec5746..7b361bc622 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java @@ -29,6 +29,7 @@ import org.junit.Assert; import org.junit.BeforeClass; import org.apache.cassandra.distributed.impl.AbstractCluster; +import org.apache.cassandra.distributed.impl.IsolatedExecutor; public class DistributedTestBase { @@ -41,15 +42,38 @@ public class DistributedTestBase public static String KEYSPACE = "distributed_test_keyspace"; + public static void nativeLibraryWorkaround() + { + // Disable the Netty tcnative library otherwise the io.netty.internal.tcnative.CertificateCallbackTask, + // CertificateVerifierTask, SSLPrivateKeyMethodDecryptTask, SSLPrivateKeyMethodSignTask, + // SSLPrivateKeyMethodTask, and SSLTask hold a gcroot against the InstanceClassLoader. + System.setProperty("cassandra.disable_tcactive_openssl", "true"); + System.setProperty("io.netty.transport.noNative", "true"); + } + + public static void processReaperWorkaround() + { + // Make sure the 'process reaper' thread is initially created under the main classloader, + // otherwise it gets created with the contextClassLoader pointing to an InstanceClassLoader + // which prevents it from being garbage collected. + IsolatedExecutor.ThrowingRunnable.toRunnable(() -> new ProcessBuilder().command("true").start().waitFor()).run(); + } + @BeforeClass public static void setup() { System.setProperty("org.apache.cassandra.disable_mbean_registration", "true"); + nativeLibraryWorkaround(); + processReaperWorkaround(); + } + + static String withKeyspace(String replaceIn) + { + return String.format(replaceIn, KEYSPACE); } protected static > C init(C cluster) { - cluster.startup(); cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + cluster.size() + "};"); return cluster; } diff --git a/test/distributed/org/apache/cassandra/distributed/test/GossipTest.java b/test/distributed/org/apache/cassandra/distributed/test/GossipTest.java index 11e9985a6c..c2e9e4fa15 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/GossipTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/GossipTest.java @@ -20,8 +20,6 @@ package org.apache.cassandra.distributed.test; import java.net.InetAddress; import java.util.Collection; -import java.util.Collections; -import java.util.EnumSet; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.LockSupport; import java.util.stream.Collectors; @@ -30,17 +28,17 @@ import com.google.common.collect.Iterables; import org.junit.Assert; import org.junit.Test; -import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.dht.Token; import org.apache.cassandra.distributed.Cluster; -import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.gms.ApplicationState; import org.apache.cassandra.gms.EndpointState; import org.apache.cassandra.gms.Gossiper; -import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + public class GossipTest extends DistributedTestBase { @@ -48,8 +46,12 @@ public class GossipTest extends DistributedTestBase public void nodeDownDuringMove() throws Throwable { int liveCount = 1; + System.setProperty("cassandra.ring_delay_ms", "5000"); // down from 30s default System.setProperty("cassandra.consistent.rangemovement", "false"); - try (Cluster cluster = Cluster.create(2 + liveCount, EnumSet.of(Feature.GOSSIP))) + System.setProperty("cassandra.consistent.simultaneousmoves.allow", "true"); + try (Cluster cluster = Cluster.build(2 + liveCount) + .withConfig(config -> config.with(NETWORK).with(GOSSIP)) + .createWithoutStarting()) { int fail = liveCount + 1; int late = fail + 1; diff --git a/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java b/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java new file mode 100644 index 0000000000..4bfbdc9c18 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java @@ -0,0 +1,202 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test; + +import java.io.File; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.nio.file.FileSystems; +import java.nio.file.Path; +import java.sql.Date; +import java.text.SimpleDateFormat; +import java.time.Instant; +import java.util.List; +import java.util.function.Consumer; +import javax.management.MBeanServer; + +import org.junit.Ignore; +import org.junit.Test; + +import com.sun.management.HotSpotDiagnosticMXBean; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.impl.InstanceConfig; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.service.CassandraDaemon; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.SigarLibrary; + +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + +/* Resource Leak Test - useful when tracking down issues with in-JVM framework cleanup. + * All objects referencing the InstanceClassLoader need to be garbage collected or + * the JVM runs out of metaspace. This test also calls out to lsof to check which + * file handles are still opened. + * + * This is intended to be a burn type test where it is run outside of the test suites + * when a problem is detected (like OutOfMetaspace exceptions). + * + * Currently this test demonstrates that the InstanceClassLoader is cleaned up (load up + * the final hprof and check that the class loaders are not reachable from a GC root), + * but it shows that the file handles for Data/Index files are being leaked. + */ +@Ignore +public class ResourceLeakTest extends DistributedTestBase +{ + // Parameters to adjust while hunting for leaks + final int numTestLoops = 1; // Set this value high to crash on leaks, or low when tracking down an issue. + final boolean dumpEveryLoop = false; // Dump heap & possibly files every loop + final boolean dumpFileHandles = false; // Call lsof whenever dumping resources + final boolean forceCollection = false; // Whether to explicitly force finalization/gc for smaller heap dumps + final long finalWaitMillis = 0l; // Number of millis to wait before final resource dump to give gc a chance + + static final SimpleDateFormat format = new SimpleDateFormat("yyyyMMddHHmmss"); + static final String when = format.format(Date.from(Instant.now())); + + static String outputFilename(String base, String description, String extension) + { + Path p = FileSystems.getDefault().getPath("build", "test", + String.join("-", when, base, description) + extension); + return p.toString(); + } + + /** + * Retrieves the process ID or null if the process ID cannot be retrieved. + * @return the process ID or null if the process ID cannot be retrieved. + * + * (Duplicated from HeapUtils to avoid refactoring older releases where this test is useful). + */ + private static Long getProcessId() + { + // Once Java 9 is ready the process API should provide a better way to get the process ID. + long pid = SigarLibrary.instance.getPid(); + + if (pid >= 0) + return Long.valueOf(pid); + + return getProcessIdFromJvmName(); + } + + /** + * Retrieves the process ID from the JVM name. + * @return the process ID or null if the process ID cannot be retrieved. + */ + private static Long getProcessIdFromJvmName() + { + // the JVM name in Oracle JVMs is: '@' but this might not be the case on all JVMs + String jvmName = ManagementFactory.getRuntimeMXBean().getName(); + try + { + return Long.parseLong(jvmName.split("@")[0]); + } + catch (NumberFormatException e) + { + // ignore + } + return null; + } + + static void dumpHeap(String description, boolean live) throws IOException + { + MBeanServer server = ManagementFactory.getPlatformMBeanServer(); + HotSpotDiagnosticMXBean mxBean = ManagementFactory.newPlatformMXBeanProxy( + server, "com.sun.management:type=HotSpotDiagnostic", HotSpotDiagnosticMXBean.class); + mxBean.dumpHeap(outputFilename("heap", description, ".hprof"), live); + } + + static void dumpOpenFiles(String description) throws IOException, InterruptedException + { + long pid = getProcessId(); + ProcessBuilder map = new ProcessBuilder("/usr/sbin/lsof", "-p", Long.toString(pid)); + File output = new File(outputFilename("lsof", description, ".txt")); + map.redirectOutput(output); + map.redirectErrorStream(true); + map.start().waitFor(); + } + + void dumpResources(String description) throws IOException, InterruptedException + { + dumpHeap(description, false); + if (dumpFileHandles) + { + dumpOpenFiles(description); + } + } + + void doTest(int numClusterNodes, Consumer updater) throws Throwable + { + for (int loop = 0; loop < numTestLoops; loop++) + { + try (Cluster cluster = Cluster.build(numClusterNodes).withConfig(updater).start()) + { + if (cluster.get(1).config().has(GOSSIP)) // Wait for gossip to settle on the seed node + cluster.get(1).runOnInstance(() -> Gossiper.waitToSettle()); + + init(cluster); + String tableName = "tbl" + loop; + cluster.schemaChange("CREATE TABLE " + KEYSPACE + "." + tableName + " (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + "." + tableName + "(pk,ck,v) VALUES (0,0,0)", ConsistencyLevel.ALL); + cluster.get(1).callOnInstance(() -> FBUtilities.waitOnFutures(Keyspace.open(KEYSPACE).flush())); + if (dumpEveryLoop) + { + dumpResources(String.format("loop%03d", loop)); + } + } + catch (Throwable tr) + { + System.out.println("Dumping resources for exception: " + tr.getMessage()); + tr.printStackTrace(); + dumpResources("exception"); + } + if (forceCollection) + { + System.runFinalization(); + System.gc(); + } + } + } + + @Test + public void looperTest() throws Throwable + { + doTest(1, config -> {}); + if (forceCollection) + { + System.runFinalization(); + System.gc(); + Thread.sleep(finalWaitMillis); + } + dumpResources("final"); + } + + @Test + public void looperGossipNetworkTest() throws Throwable + { + doTest(2, config -> config.with(GOSSIP).with(NETWORK)); + if (forceCollection) + { + System.runFinalization(); + System.gc(); + Thread.sleep(finalWaitMillis); + } + dumpResources("final-gossip-network"); + } +} diff --git a/test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java b/test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java index e4c695cdf0..8526dd04fd 100644 --- a/test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java +++ b/test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java @@ -22,6 +22,7 @@ import java.io.OutputStream; import java.io.PrintStream; import java.util.Arrays; import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; import org.junit.Assert; import org.junit.Test; @@ -56,7 +57,7 @@ public class SEPExecutorTest } // shutdown does not guarantee that threads are actually dead once it exits, only that they will stop promptly afterwards - sharedPool.shutdownAndWait(); + sharedPool.shutdownAndWait(1L, TimeUnit.MINUTES); for (Thread thread : Thread.getAllStackTraces().keySet()) { if (thread.getName().contains(MAGIC))