Merge branch 'cassandra-3.0' into cassandra-3.11

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

build.xml

@@ -1852,10 +1852,24 @@
       </jar>
   </target>
 
-  <target name="test-jvm-dtest" depends="build-test" description="Execute unit tests">
+  <target name="test-jvm-dtest" depends="build-test" description="Execute in-jvm dtests">
     <testmacro inputdir="${test.distributed.src}" timeout="${test.distributed.timeout}" forkmode="once" showoutput="true" filter="**/test/*Test.java">
       <jvmarg value="-Dlogback.configurationFile=test/conf/logback-dtest.xml"/>
-      <jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
+      <jvmarg value="-Dcassandra.ring_delay_ms=10000"/>
+      <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
+      <jvmarg value="-Djava.io.tmpdir=${tmp.dir}"/>
+      <jvmarg value="-Dcassandra.skip_sync=true" />
+    </testmacro>
+  </target>
+
+  <!-- Use this with an FQDN for test class, and a csv list of methods like this:
+      ant test-jvm-dtest-some -Dtest.name=org.apache.cassandra.distributed.test.ResourceLeakTest -Dtest.methods=looperTest
+    -->
+  <target name="test-jvm-dtest-some" depends="build-test" description="Execute some in-jvm dtests">
+    <testmacro inputdir="${test.distributed.src}" timeout="${test.distributed.timeout}" forkmode="once" showoutput="true">
+      <test name="${test.name}" methods="${test.methods}" outfile="build/test/output/TEST-${test.name}-${test.methods}"/>
+      <jvmarg value="-Dlogback.configurationFile=test/conf/logback-dtest.xml"/>
+      <jvmarg value="-Dcassandra.ring_delay_ms=10000"/>
       <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
       <jvmarg value="-Djava.io.tmpdir=${tmp.dir}"/>
       <jvmarg value="-Dcassandra.skip_sync=true" />

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();
+        ExecutorUtils.shutdownAndWait(timeout, unit, batchlogTasks);
-        batchlogTasks.awaitTermination(60, TimeUnit.SECONDS);
     }
 
     public static void remove(UUID id)

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

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 };
+        ExecutorUtils.shutdownNowAndWait(timeout, unit, scheduledFastTasks, scheduledTasks, nonPeriodicTasks, optionalTasks);
-        for (ExecutorService executor : executors)
-            executor.shutdownNow();
-        for (ExecutorService executor : executors)
-            executor.awaitTermination(60, TimeUnit.SECONDS);
     }
 }

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

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())
+        ExecutorUtils.shutdownNowAndWait(timeout, unit, StageManager.stages.values());
-            StageManager.stages.get(stage).shutdownNow();
-        for (Stage stage : Stage.values())
-            StageManager.stages.get(stage).awaitTermination(60, TimeUnit.SECONDS);
     }
 
     /**

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();
+        List<ExecutorService> executors = new ArrayList<>(perDiskflushExecutors.length + 3);
-        reclaimExecutor.awaitTermination(60, TimeUnit.SECONDS);
+        Collections.addAll(executors, reclaimExecutor, postFlushExecutor, flushExecutor);
-    }
+        Collections.addAll(executors, perDiskflushExecutors);
-
+        ExecutorUtils.shutdownAndWait(timeout, unit, executors);
-    public static void shutdownPerDiskFlushExecutors() throws InterruptedException
-    {
-        for (ExecutorService executorService : perDiskflushExecutors)
-            executorService.shutdown();
-        for (ExecutorService executorService : perDiskflushExecutors)
-            executorService.awaitTermination(60, TimeUnit.SECONDS);
     }
 
     public void reload()

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
     {

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

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<String, Object> writerParams)
     {
-        try
+        try(Stream<Path> list = Files.list(hintsDirectory.toPath()))
         {
             Map<UUID, List<HintsDescriptor>> stores =
-                Files.list(hintsDirectory.toPath())
+                     list
                      .filter(HintsDescriptor::isHintFileName)
                      .map(HintsDescriptor::readFromFileQuietly)
                      .filter(Optional::isPresent)

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)
+        shutdown(executors);
-            executor.shutdown();
+        awaitTermination(timeout, unit, executors);
-        for (ExecutorService executor : executors)
-            executor.awaitTermination(60, TimeUnit.SECONDS);
     }
 }

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

src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java

@@ -2371,10 +2371,10 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
 
     }
 
-    public static void shutdownBlocking() throws InterruptedException
+    public static void shutdownBlocking(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
     {
-        syncExecutor.shutdownNow();
+
-        syncExecutor.awaitTermination(0, TimeUnit.SECONDS);
+        ExecutorUtils.shutdownNowAndWait(timeout, unit, syncExecutor);
         resetTidying();
     }
 }

src/java/org/apache/cassandra/net/MessagingService.java

@@ -1383,6 +1383,8 @@ public final class MessagingService implements MessagingServiceMBean
             {
                 case "Unknown error: 316":
                 case "No such file or directory":
+                case "Bad file descriptor":
+                case "Thread signal failed":
                     return;
             }
         }

src/java/org/apache/cassandra/net/OutboundTcpConnection.java

@@ -426,7 +426,7 @@ public class OutboundTcpConnection extends FastThreadLocalThread
 
         long start = System.nanoTime();
         long timeout = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getRpcTimeout());
-        while (System.nanoTime() - start < timeout)
+        while (System.nanoTime() - start < timeout && !isStopped)
         {
             targetVersion = MessagingService.instance().getVersion(poolReference.endPoint());
             try

src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java

@@ -23,15 +23,21 @@ import org.apache.cassandra.concurrent.NamedThreadFactory;
 import org.apache.cassandra.config.Schema;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
+import org.apache.cassandra.utils.ExecutorUtils;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import com.google.common.annotations.VisibleForTesting;
 
+import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination;
+import static org.apache.cassandra.utils.ExecutorUtils.shutdownNow;
+
 public class PendingRangeCalculatorService
 {
     public static final PendingRangeCalculatorService instance = new PendingRangeCalculatorService();
@@ -112,9 +118,8 @@ public class PendingRangeCalculatorService
     }
 
     @VisibleForTesting
-    public void shutdownExecutor() throws InterruptedException
+    public void shutdownExecutor(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
     {
-        executor.shutdown();
+        ExecutorUtils.shutdownNowAndWait(timeout, unit, executor);
-        executor.awaitTermination(60, TimeUnit.SECONDS);
     }
 }

src/java/org/apache/cassandra/service/StorageService.java

@@ -18,7 +18,6 @@
 package org.apache.cassandra.service;
 
 import java.io.*;
-import java.lang.management.ManagementFactory;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
@@ -98,6 +97,7 @@ import org.apache.cassandra.utils.progress.jmx.JMXProgressSupport;
 import org.apache.cassandra.utils.progress.jmx.LegacyJMXProgressSupport;
 
 import static java.util.Arrays.asList;
+import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.stream.Collectors.toList;
 import static org.apache.cassandra.index.SecondaryIndexManager.getIndexName;
 import static org.apache.cassandra.index.SecondaryIndexManager.isIndexColumnFamily;
@@ -4560,7 +4560,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         {
             setMode(Mode.DRAINING, "starting drain process", !isFinalShutdown);
 
-            BatchlogManager.instance.shutdown();
+            try
+            {
+                /* not clear this is reasonable time, but propagated from prior embedded behaviour */
+                BatchlogManager.instance.shutdownAndWait(1L, MINUTES);
+            }
+            catch (TimeoutException t)
+            {
+                logger.error("Batchlog manager timed out shutting down", t);
+            }
+
             HintsService.instance.pauseDispatch();
 
             if (daemon != null)
@@ -4653,7 +4662,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
 
             // wait for miscellaneous tasks like sstable and commitlog segment deletion
             ScheduledExecutors.nonPeriodicTasks.shutdown();
-            if (!ScheduledExecutors.nonPeriodicTasks.awaitTermination(1, TimeUnit.MINUTES))
+            if (!ScheduledExecutors.nonPeriodicTasks.awaitTermination(1, MINUTES))
                 logger.warn("Failed to wait for non periodic tasks to shutdown");
 
             ColumnFamilyStore.shutdownPostFlushExecutor();
@@ -5286,4 +5295,13 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         DatabaseDescriptor.setHintedHandoffThrottleInKB(throttleInKB);
         logger.info("Updated hinted_handoff_throttle_in_kb to {}", throttleInKB);
     }
+
+    @VisibleForTesting
+    public void shutdownServer()
+    {
+        if (drainOnShutdown != null)
+        {
+            Runtime.getRuntime().removeShutdownHook(drainOnShutdown);
+        }
+    }
 }

src/java/org/apache/cassandra/streaming/StreamCoordinator.java

@@ -19,11 +19,17 @@ package org.apache.cassandra.streaming;
 
 import java.net.InetAddress;
 import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
+import org.apache.cassandra.utils.ExecutorUtils;
 import org.apache.cassandra.utils.FBUtilities;
 
 /**
@@ -341,4 +347,11 @@ public class StreamCoordinator
             return sessionInfos.values();
         }
     }
+
+    @VisibleForTesting
+    public static void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
+    {
+        ExecutorUtils.shutdownAndWait(timeout, unit, streamExecutor);
+    }
+
 }

src/java/org/apache/cassandra/streaming/StreamSession.java

@@ -48,6 +48,7 @@ import org.apache.cassandra.metrics.StreamingMetrics;
 import org.apache.cassandra.service.ActiveRepairService;
 import org.apache.cassandra.streaming.messages.*;
 import org.apache.cassandra.utils.CassandraVersion;
+import org.apache.cassandra.utils.ExecutorUtils;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.Pair;
@@ -837,4 +838,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber
             }
         }
     }
+
+    @VisibleForTesting
+    public static void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
+    {
+        List<ExecutorService> executors = ImmutableList.of(keepAliveExecutor);
+        ExecutorUtils.shutdownNow(executors);
+        ExecutorUtils.awaitTermination(timeout, unit, executors);
+    }
 }

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

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<T> implements RefCounted<T>
     }
 
     @VisibleForTesting
-    public static void shutdownReferenceReaper() throws InterruptedException
+    public static void shutdownReferenceReaper(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
     {
-        EXEC.shutdown();
+        ExecutorUtils.shutdownNowAndWait(timeout, unit, EXEC, STRONG_LEAK_DETECTOR);
-        EXEC.awaitTermination(60, TimeUnit.SECONDS);
-        if (STRONG_LEAK_DETECTOR != null)
-        {
-            STRONG_LEAK_DETECTOR.shutdownNow();
-            STRONG_LEAK_DETECTOR.awaitTermination(60, TimeUnit.SECONDS);
-        }
     }
 }

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();
+        shutdownNow(Arrays.asList(EXEC));
-        EXEC.awaitTermination(60, TimeUnit.SECONDS);
+        awaitTermination(timeout, unit, Arrays.asList(EXEC));
     }
 }

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();
+        ExecutorUtils.shutdownNowAndWait(timeout, unit, cleaner);
-        cleaner.awaitTermination(60, TimeUnit.SECONDS);
     }
 
+
     public abstract MemtableAllocator newAllocator();
 
     /**

test/conf/logback-dtest.xml

@@ -23,7 +23,7 @@
   <!-- Shutdown hook ensures that async appender flushes -->
   <shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/>
 
-  <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+  <appender name="INSTANCEFILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
 
     <file>./build/test/logs/${cassandra.testtag}/TEST-${suitename}.log</file>
     <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
@@ -42,14 +42,14 @@
     <immediateFlush>false</immediateFlush>
   </appender>
 
-  <appender name="ASYNCFILE" class="ch.qos.logback.classic.AsyncAppender">
+  <appender name="INSTANCEASYNCFILE" class="ch.qos.logback.classic.AsyncAppender">
     <discardingThreshold>0</discardingThreshold>
     <maxFlushTime>0</maxFlushTime>
     <queueSize>1024</queueSize>
-    <appender-ref ref="FILE"/>
+    <appender-ref ref="INSTANCEFILE"/>
   </appender>
 
-  <appender name="STDERR" target="System.err" class="ch.qos.logback.core.ConsoleAppender">
+  <appender name="INSTANCESTDERR" target="System.err" class="ch.qos.logback.core.ConsoleAppender">
     <encoder>
       <pattern>%-5level %date{HH:mm:ss,SSS} %msg%n</pattern>
     </encoder>
@@ -58,7 +58,7 @@
     </filter>
   </appender>
 
-  <appender name="STDOUT" target="System.out" class="ch.qos.logback.core.ConsoleAppender">
+  <appender name="INSTANCESTDOUT" target="System.out" class="ch.qos.logback.core.ConsoleAppender">
     <encoder>
       <pattern>%-5level %date{HH:mm:ss,SSS} %msg%n</pattern>
     </encoder>
@@ -67,7 +67,7 @@
     </filter>
   </appender>
 
-  <appender name="STDOUT" target="System.out" class="ch.qos.logback.core.ConsoleAppender">
+  <appender name="INSTANCESTDOUT" target="System.out" class="ch.qos.logback.core.ConsoleAppender">
     <encoder>
       <pattern>%-5level [%thread] ${instance_id} %date{ISO8601} %F:%L - %msg%n</pattern>
     </encoder>
@@ -79,8 +79,8 @@
   <logger name="org.apache.hadoop" level="WARN"/>
 
   <root level="DEBUG">
-    <appender-ref ref="ASYNCFILE" />
+    <appender-ref ref="INSTANCEASYNCFILE" />
-    <appender-ref ref="STDERR" />
+    <appender-ref ref="INSTANCESTDERR" />
-    <appender-ref ref="STDOUT" />
+    <appender-ref ref="INSTANCESTDOUT" />
   </root>
 </configuration>

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<IInvokableInstance> implements ICluster, AutoCloseable
 {
-    private Cluster(File root, Versions.Version version, List<InstanceConfig> configs, Set<Feature> features, ClassLoader sharedClassLoader)
+    private Cluster(File root, Versions.Version version, List<InstanceConfig> 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<IInvokableInstance, Cluster> build(int nodeCount)
+    {
+        return new Builder<>(nodeCount, Cluster::new);
+    }
+
+    public static Cluster create(int nodeCount, Consumer<InstanceConfig> configUpdater) throws IOException
+    {
+        return build(nodeCount).withConfig(configUpdater).start();
     }
 
     public static Cluster create(int nodeCount) throws Throwable
     {
-        return create(nodeCount, Cluster::new);
+        return build(nodeCount).start();
-    }
-    public static Cluster create(int nodeCount, Set<Feature> 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<Feature> with)
-    {
-        return create(nodeCount, Versions.CURRENT, root, with, Cluster::new);
     }
 }
 

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<IUpgradeableInstance> implements ICluster, AutoCloseable
 {
-    private UpgradeableCluster(File root, Versions.Version version, List<InstanceConfig> configs, Set<Feature> features, ClassLoader sharedClassLoader)
+    private UpgradeableCluster(File root, Versions.Version version, List<InstanceConfig> 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<IUpgradeableInstance, UpgradeableCluster> 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);
-    }
-
 }
 

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<Void> 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<Feature> with);
+    void startup(ICluster cluster);
     void receiveMessage(IMessage message);
 
     int getMessagingVersion();

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

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<InetAddressAndPort, IMessage> filter(BiConsumer<InetAddressAndPort, IMessage> applyIfNotFiltered);
+    boolean permit(IInstance from, IInstance to, int verb);
+
 }

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<I extends IInstance> 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<Feature> features;
 
     // mutated by starting/stopping a node
     private final List<I> instances;
@@ -103,6 +103,7 @@ public abstract class AbstractCluster<I extends IInstance> 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<I extends IInstance> 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<IInstanceConfig, ClassLoader, Instance>)Instance::new, classLoader)
                            .apply(config, classLoader);
         }
@@ -145,11 +147,17 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
         {
             if (!isShutdown)
                 throw new IllegalStateException();
-            delegate().startup(AbstractCluster.this, features);
+            delegate().startup(AbstractCluster.this);
             isShutdown = false;
             updateMessagingVersions();
         }
 
+        @Override
+        public synchronized Future<Void> shutdown()
+        {
+            return shutdown(true);
+        }
+
         @Override
         public synchronized Future<Void> shutdown(boolean graceful)
         {
@@ -185,26 +193,26 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
         }
     }
 
-    protected AbstractCluster(File root, Versions.Version version, List<InstanceConfig> configs, Set<Feature> features, ClassLoader sharedClassLoader)
+    protected AbstractCluster(File root, Versions.Version version, List<InstanceConfig> 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<I extends IInstance> 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<I extends IInstance> implements ICluster,
 
     protected interface Factory<I extends IInstance, C extends AbstractCluster<I>>
     {
-        C newCluster(File root, Versions.Version version, List<InstanceConfig> configs, Set<Feature> features, ClassLoader sharedClassLoader);
+        C newCluster(File root, Versions.Version version, List<InstanceConfig> configs, ClassLoader sharedClassLoader);
     }
 
-    protected static <I extends IInstance, C extends AbstractCluster<I>> C
+    public static class Builder<I extends IInstance, C extends AbstractCluster<I>>
-    create(int nodeCount, Factory<I, C> factory) throws Throwable
     {
-        return create(nodeCount, Collections.emptySet(), factory);
+        private final int nodeCount;
-    }
+        private final Factory<I, C> factory;
-
+        private int subnet;
-    protected static <I extends IInstance, C extends AbstractCluster<I>> C
+        private File root;
-    create(int nodeCount, Set<Feature> features, Factory<I, C> factory) throws Throwable
+        private Versions.Version version;
-    {
+        private Consumer<InstanceConfig> configUpdater;
-        return create(nodeCount, Files.createTempDirectory("dtests").toFile(), features, factory);
+        public Builder(int nodeCount, Factory<I, C> factory)
-    }
-
-    protected static <I extends IInstance, C extends AbstractCluster<I>> C
-    create(int nodeCount, File root, Factory<I, C> factory)
-    {
-        return create(nodeCount, root, Collections.emptySet(), factory);
-    }
-
-    protected static <I extends IInstance, C extends AbstractCluster<I>> C
-    create(int nodeCount, File root, Set<Feature> features, Factory<I, C> factory)
-    {
-        return create(nodeCount, Versions.CURRENT, root, features, factory);
-    }
-
-    protected static <I extends IInstance, C extends AbstractCluster<I>> C
-    create(int nodeCount, Versions.Version version, Factory<I, C> factory) throws IOException
-    {
-        return create(nodeCount, version, Collections.emptySet(), factory);
-    }
-
-    protected static <I extends IInstance, C extends AbstractCluster<I>> C
-    create(int nodeCount, Versions.Version version, Set<Feature> features, Factory<I, C> factory) throws IOException
-    {
-        return create(nodeCount, version, Files.createTempDirectory("dtests").toFile(), features, factory);
-    }
-
-    protected static <I extends IInstance, C extends AbstractCluster<I>> C
-    create(int nodeCount, Versions.Version version, File root, Factory<I, C> factory)
-    {
-        return create(nodeCount, version, root, Collections.emptySet(), factory);
-    }
-    protected static <I extends IInstance, C extends AbstractCluster<I>> C
-    create(int nodeCount, Versions.Version version, File root, Set<Feature> features, Factory<I, C> factory)
-    {
-        root.mkdirs();
-        setupLogging(root);
-
-        ClassLoader sharedClassLoader = Thread.currentThread().getContextClassLoader();
-
-        List<InstanceConfig> configs = new ArrayList<>();
-        long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount);
-        for (int i = 0 ; i < nodeCount ; ++i)
         {
-            InstanceConfig config = InstanceConfig.generate(i + 1, root, String.valueOf(token));
+            this.nodeCount = nodeCount;
-            configs.add(config);
+            this.factory = factory;
-            token += increment;
         }
 
-        C cluster = factory.newCluster(root, version, configs, features, sharedClassLoader);
+        public Builder<I, C> withSubnet(int subnet)
-        return cluster;
+        {
+            this.subnet = subnet;
+            return this;
+        }
+
+        public Builder<I, C> withRoot(File root)
+        {
+            this.root = root;
+            return this;
+        }
+
+        public Builder<I, C> withVersion(Versions.Version version)
+        {
+            this.version = version;
+            return this;
+        }
+
+        public Builder<I, C> withConfig(Consumer<InstanceConfig> 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<InstanceConfig> configs = new ArrayList<>();
+            long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount);
+            for (int i = 0; i < nodeCount; ++i)
+            {
+                InstanceConfig config = InstanceConfig.generate(i + 1, 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

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<Feature> with)
+    public void startup(ICluster cluster)
     {
-        delegate().startup(cluster, with);
+        delegate().startup(cluster);
     }
 
     @Override

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) -> {
+        // Set the config at instance creation, possibly before startup() has run on all other instances.
-            Config.setOverrideLoadConfig(() -> loadConfig(override));
+        // setMessagingVersions below will call runOnInstance which will instantiate
-            DatabaseDescriptor.daemonInitialization();
+        // the MessagingService and dependencies preventing later changes to network parameters.
-        }).accept(config);
+        Config.setOverrideLoadConfig(() -> loadConfig(config));
     }
 
     public IInstanceConfig config()
@@ -182,7 +188,10 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
     private void registerMockMessaging(ICluster cluster)
     {
         BiConsumer<InetAddressAndPort, IMessage> deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message);
-        BiConsumer<InetAddressAndPort, IMessage> deliverToInstanceIfNotFiltered = cluster.filters().filter(deliverToInstance);
+        BiConsumer<InetAddressAndPort, IMessage> deliverToInstanceIfNotFiltered = (to, message) -> {
+            if (cluster.filters().permit(this, cluster.get(to), message.verb()))
+                deliverToInstance.accept(to, message);
+        };
 
         Map<InetAddress, InetAddressAndPort> 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<InetAddressAndPort, IMessage> 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());
+                int version = imessage.version();
-                Runnable deliver = new MessageDeliveryTask(messageIn, message.id());
+
-                deliver.run();
+                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<Feature> 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);
                 }
-
+                if (config.has(NETWORK))
-                // TODO: support each separately
-                if (with.contains(Feature.GOSSIP) || with.contains(Feature.NETWORK))
                 {
-                    StorageService.instance.prepareToJoin();
+                    registerFilter(cluster);
-                    StorageService.instance.joinTokenRing(5000);
+                    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,
+                                () -> Gossiper.instance.stopShutdownAndWait(1L, MINUTES),
-                    CompactionManager.instance::forceShutdown,
+                                CompactionManager.instance::forceShutdown,
-                    BatchlogManager.instance::shutdown,
+                                () -> BatchlogManager.instance.shutdownAndWait(1L, MINUTES),
-                    HintsService.instance::shutdownBlocking,
+                                HintsService.instance::shutdownBlocking,
-                    SecondaryIndexManager::shutdownExecutors,
+                                () -> StreamCoordinator.shutdownAndWait(1L, MINUTES),
-                    ColumnFamilyStore::shutdownFlushExecutor,
+                                () -> StreamSession.shutdownAndWait(1L, MINUTES),
-                    ColumnFamilyStore::shutdownPostFlushExecutor,
+                                () -> SecondaryIndexManager.shutdownAndWait(1L, MINUTES),
-                    ColumnFamilyStore::shutdownReclaimExecutor,
+                                () -> IndexSummaryManager.instance.shutdownAndWait(1L, MINUTES),
-                    ColumnFamilyStore::shutdownPerDiskFlushExecutors,
+                                () -> ColumnFamilyStore.shutdownExecutorsAndWait(1L, MINUTES),
-                    PendingRangeCalculatorService.instance::shutdownExecutor,
+                                () -> PendingRangeCalculatorService.instance.shutdownExecutor(1L, MINUTES),
-                    BufferPool::shutdownLocalCleaner,
+                                () -> BufferPool.shutdownLocalCleaner(1L, MINUTES),
-                    Ref::shutdownReferenceReaper,
+                                () -> Ref.shutdownReferenceReaper(1L, MINUTES),
-                    Memtable.MEMORY_POOL::shutdown,
+                                () -> Memtable.MEMORY_POOL.shutdownAndWait(1L, MINUTES),
-                    ScheduledExecutors::shutdownAndWait,
+                                () -> SSTableReader.shutdownBlocking(1L, MINUTES)
-                    SSTableReader::shutdownBlocking
             );
             error = parallelRun(error, executor,
+                                () -> ScheduledExecutors.shutdownAndWait(1L, MINUTES),
                                 MessagingService.instance()::shutdown
             );
             error = parallelRun(error, executor,
-                                StageManager::shutdownAndWait,
+                                () -> StageManager.shutdownAndWait(1L, MINUTES),
-                                SharedExecutorPool.SHARED::shutdownAndWait
+                                () -> SharedExecutorPool.SHARED.shutdownAndWait(1L, MINUTES)
             );
             error = parallelRun(error, executor,
                                 CommitLog.instance::shutdownBlocking
             );
 
-            LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
-            loggerContext.stop();
             Throwables.maybeFail(error);
         }).apply(isolatedExecutor);
 

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) +
                '}';
     }

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<String, Object> 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,
+                                  ipPrefix + nodeNum,
-                                  "127.0.0." + nodeNum,
+                                  ipPrefix + nodeNum,
-                                  "127.0.0." + nodeNum,
+                                  ipPrefix + nodeNum,
-                                  "127.0.0." + 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),

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<Void> shutdown()
     {
         isolatedExecutor.shutdown();
-        ThrowingRunnable.toRunnable(((URLClassLoader) classLoader)::close).run();
+
-        return CompletableFuture.runAsync(ThrowingRunnable.toRunnable(() -> isolatedExecutor.awaitTermination(60, TimeUnit.SECONDS)),
+        /* Use a thread pool with a core pool size of zero to terminate the thread as soon as possible
-                                          Executors.newSingleThreadExecutor());
+        ** 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<Runnable>(), 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 <O> CallableNoExcept<Future<O>> async(CallableNoExcept<O> call) { return () -> isolatedExecutor.submit(call); }

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<Filter> 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<InetAddressAndPort, IMessage> filter(BiConsumer<InetAddressAndPort, IMessage> applyIfNotFiltered)
+        for (Filter filter : filters)
-    {
+            if (filter.matches(fromNum, toNum, verb))
-        return (toAddress, message) ->
+                return false;
-        {
-            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;
-            }
 
-            applyIfNotFiltered.accept(toAddress, message);
+        return true;
-        };
     }
 
     public class Filter implements IMessageFilters.Filter

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 extends AbstractCluster<?>> C init(C cluster)
     {
-        cluster.startup();
         cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + cluster.size() + "};");
         return cluster;
     }

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;

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 <code>null</code> if the process ID cannot be retrieved.
+     * @return the process ID or <code>null</code> 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 <code>null</code> if the process ID cannot be retrieved.
+     */
+    private static Long getProcessIdFromJvmName()
+    {
+        // the JVM name in Oracle JVMs is: '<pid>@<hostname>' 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<InstanceConfig> 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");
+    }
+}

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