Merge branch 'cassandra-3.0' into cassandra-3.11

.circleci/config.yml

@@ -172,7 +172,7 @@ jobs:
             # get all of our unit test filenames
             set -eo pipefail && circleci tests glob "$HOME/cassandra/test/unit/**/*.java" > /tmp/all_java_unit_tests.txt
             # append distributed tests
-            set -eo pipefail && circleci tests glob "$HOME/cassandra/test/distributed/**/*.java" > /tmp/all_java_distributed_tests.txt
+            set -eo pipefail && circleci tests glob "$HOME/cassandra/test/distributed/**/test/*.java" > /tmp/all_java_distributed_tests.txt
 
             # split up the unit tests into groups based on the number of containers we have
             set -eo pipefail && circleci tests split --split-by=timings --timings-type=filename --index=${CIRCLE_NODE_INDEX} --total=${CIRCLE_NODE_TOTAL} /tmp/all_java_unit_tests.txt > /tmp/java_tests_${CIRCLE_NODE_INDEX}.txt

CHANGES.txt

@@ -1,8 +1,3 @@
-3.11.5
-Merged from 3.0:
- * Catch CorruptSSTableExceptions and FSErrors in ALAExecutorService (CASSANDRA-14993)
-
-
 3.11.4
  * Make stop-server.bat wait for Cassandra to terminate (CASSANDRA-14829)
  * Correct sstable sorting for garbagecollect and levelled compaction (CASSANDRA-14870)

build.xml

@@ -1223,6 +1223,15 @@
     </copy>
   </target>
 
+  <target name="dtest-jar" depends="build-test, build" description="Create dtest-compatible jar, including all dependencies">
+      <jar jarfile="${build.dir}/dtest-${base.version}.jar">
+          <zipgroupfileset dir="${build.lib}" includes="*.jar" excludes="META-INF/*.SF"/>
+          <fileset dir="${build.classes.main}"/>
+          <fileset dir="${test.classes}"/>
+          <fileset dir="${test.conf}" />
+      </jar>
+  </target>
+
   <!-- Defines how to run a set of tests. If you change the defaults for attributes
        you should also update them in testmacro.,
        The two are split because the helper doesn't generate
@@ -1794,7 +1803,7 @@
 
   <target name="test" depends="build-test,get-cores,get-mem,stress-build" description="Parallel Test Runner">
     <path id="all-test-classes-path">
-      <fileset dir="${test.unit.src}" includes="**/${test.name}.java" />
+      <fileset dir="${test.unit.src}" includes="**/${test.name}.java" excludes="**/distributed/test/UpgradeTest*.java" />
     </path>
     <property name="all-test-classes" refid="all-test-classes-path"/>
     <testparallel testdelegate="testlist"/>

src/java/org/apache/cassandra/concurrent/AbstractLocalAwareExecutorService.java

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

src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java

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

src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java

@@ -33,6 +33,9 @@ import io.netty.util.concurrent.FastThreadLocalThread;
 
 public class NamedThreadFactory implements ThreadFactory
 {
+    private static volatile String globalPrefix;
+    public static void setGlobalPrefix(String prefix) { globalPrefix = prefix; }
+
     public final String id;
     private final int priority;
     private final ClassLoader contextClassLoader;
@@ -112,7 +115,8 @@ public class NamedThreadFactory implements ThreadFactory
 
     public static Thread createThread(ThreadGroup threadGroup, Runnable runnable, String name, boolean daemon)
     {
-        Thread thread = new FastThreadLocalThread(threadGroup, threadLocalDeallocator(runnable), name);
+        String prefix = globalPrefix;
+        Thread thread = new FastThreadLocalThread(threadGroup, threadLocalDeallocator(runnable), prefix != null ? prefix + name : name);
         thread.setDaemon(daemon);
         return thread;
     }

src/java/org/apache/cassandra/config/Config.java

@@ -25,6 +25,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 
 import com.google.common.base.Joiner;
 import com.google.common.collect.Sets;
@@ -263,6 +264,7 @@ public class Config
     public volatile int counter_cache_keys_to_save = Integer.MAX_VALUE;
 
     private static boolean isClientMode = false;
+    private static Supplier<Config> overrideLoadConfig = null;
 
     public Integer file_cache_size_in_mb;
 
@@ -398,6 +400,16 @@ public class Config
         isClientMode = clientMode;
     }
 
+    public static Supplier<Config> getOverrideLoadConfig()
+    {
+        return overrideLoadConfig;
+    }
+
+    public static void setOverrideLoadConfig(Supplier<Config> loadConfig)
+    {
+        overrideLoadConfig = loadConfig;
+    }
+
     public enum CommitLogSync
     {
         periodic,

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -261,6 +261,9 @@ public class DatabaseDescriptor
     @VisibleForTesting
     public static Config loadConfig() throws ConfigurationException
     {
+        if (Config.getOverrideLoadConfig() != null)
+            return Config.getOverrideLoadConfig().get();
+
         String loaderClass = System.getProperty(Config.PROPERTY_PREFIX + "config.loader");
         ConfigurationLoader loader = loaderClass == null
                                    ? new YamlConfigurationLoader()

src/java/org/apache/cassandra/utils/FBUtilities.java

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

test/conf/logback-dtest.xml

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

test/distributed/org/apache/cassandra/distributed/Cluster.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.List;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.IInvokableInstance;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.distributed.impl.Versions;
+
+/**
+ * A simple cluster supporting only the 'current' Cassandra version, offering easy access to the convenience methods
+ * of IInvokableInstance on each node.
+ */
+public class Cluster extends AbstractCluster<IInvokableInstance> implements ICluster, AutoCloseable
+{
+    private Cluster(File root, Versions.Version version, List<InstanceConfig> configs, ClassLoader sharedClassLoader)
+    {
+        super(root, version, configs, sharedClassLoader);
+    }
+
+    protected IInvokableInstance newInstanceWrapper(Versions.Version version, InstanceConfig config)
+    {
+        return new Wrapper(version, config);
+    }
+
+    public static Cluster create(int nodeCount) throws Throwable
+    {
+        return create(nodeCount, Cluster::new);
+    }
+    public static Cluster create(int nodeCount, File root)
+    {
+        return create(nodeCount, Versions.CURRENT, root, Cluster::new);
+    }
+}
+

test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java

@@ -1,109 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.distributed;
-
-import com.google.common.base.Predicate;
-import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.utils.Pair;
-
-import java.net.URL;
-import java.net.URLClassLoader;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.function.IntFunction;
-
-public class InstanceClassLoader extends URLClassLoader
-{
-    // Classes that have to be shared between instances, for configuration or returning values
-    private final static Class<?>[] commonClasses = new Class[]
-            {
-                    Pair.class,
-                    InstanceConfig.class,
-                    Message.class,
-                    InetAddressAndPort.class,
-                    InvokableInstance.SerializableBiConsumer.class,
-                    InvokableInstance.SerializableBiFunction.class,
-                    InvokableInstance.SerializableCallable.class,
-                    InvokableInstance.SerializableConsumer.class,
-                    InvokableInstance.SerializableFunction.class,
-                    InvokableInstance.SerializableRunnable.class,
-                    InvokableInstance.SerializableTriFunction.class,
-                    InvokableInstance.InstanceFunction.class
-            };
-
-    private final int id; // for debug purposes
-    private final ClassLoader commonClassLoader;
-    private final Predicate<String> isCommonClassName;
-
-    InstanceClassLoader(int id, URL[] urls, Predicate<String> isCommonClassName, ClassLoader commonClassLoader)
-    {
-        super(urls, null);
-        this.id = id;
-        this.commonClassLoader = commonClassLoader;
-        this.isCommonClassName = isCommonClassName;
-    }
-
-    @Override
-    public Class<?> loadClass(String name) throws ClassNotFoundException
-    {
-        // Do not share:
-        //  * yaml, which  is a rare exception because it does mess with loading org.cassandra...Config class instances
-        //  * most of the rest of Cassandra classes (unless they were explicitly shared) g
-        if (name.startsWith("org.slf4j") ||
-                name.startsWith("ch.qos.logback") ||
-                name.startsWith("org.yaml") ||
-                (name.startsWith("org.apache.cassandra") && !isCommonClassName.apply(name)))
-            return loadClassInternal(name);
-
-        return commonClassLoader.loadClass(name);
-    }
-
-    Class<?> loadClassInternal(String name) throws ClassNotFoundException
-    {
-        synchronized (getClassLoadingLock(name))
-        {
-            // First, check if the class has already been loaded
-            Class<?> c = findLoadedClass(name);
-
-            if (c == null)
-                c = findClass(name);
-
-            return c;
-        }
-    }
-
-    public static IntFunction<ClassLoader> createFactory(URLClassLoader contextClassLoader)
-    {
-        Set<String> commonClassNames = new HashSet<>();
-        for (Class<?> k : commonClasses)
-            commonClassNames.add(k.getName());
-
-        URL[] urls = contextClassLoader.getURLs();
-        return id -> new InstanceClassLoader(id, urls, commonClassNames::contains, contextClassLoader);
-    }
-
-    /**
-     * @return true iff this class was loaded by an InstanceClassLoader, and as such is used by a dtest node
-     */
-    public static boolean wasLoadedByAnInstanceClassLoader(Class<?> clazz)
-    {
-        return clazz.getClassLoader().getClass().getName().equals(InstanceClassLoader.class.getName());
-    }
-
-}

test/distributed/org/apache/cassandra/distributed/InstanceConfig.java

@@ -1,87 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.distributed;
-
-import java.io.File;
-import java.io.Serializable;
-import java.util.UUID;
-
-public class InstanceConfig implements Serializable
-{
-    public final int num;
-    public final UUID hostId =java.util.UUID.randomUUID();
-    public final String partitioner = "org.apache.cassandra.dht.Murmur3Partitioner";
-    public final String broadcast_address;
-    public final String listen_address;
-    public final String broadcast_rpc_address;
-    public final String rpc_address;
-    public final String saved_caches_directory;
-    public final String[] data_file_directories;
-    public final String commitlog_directory;
-    public final String hints_directory;
-    public final String cdc_directory;
-    public final int concurrent_writes = 2;
-    public final int concurrent_counter_writes = 2;
-    public final int concurrent_materialized_view_writes = 2;
-    public final int concurrent_reads = 2;
-    public final int memtable_flush_writers = 1;
-    public final int concurrent_compactors = 1;
-    public final int memtable_heap_space_in_mb = 10;
-    public final String initial_token;
-
-    private InstanceConfig(int num,
-                           String broadcast_address,
-                           String listen_address,
-                           String broadcast_rpc_address,
-                           String rpc_address,
-                           String saved_caches_directory,
-                           String[] data_file_directories,
-                           String commitlog_directory,
-                           String hints_directory,
-                           String cdc_directory,
-                           String initial_token)
-    {
-        this.num = num;
-        this.broadcast_address = broadcast_address;
-        this.listen_address = listen_address;
-        this.broadcast_rpc_address = broadcast_rpc_address;
-        this.rpc_address = rpc_address;
-        this.saved_caches_directory = saved_caches_directory;
-        this.data_file_directories = data_file_directories;
-        this.commitlog_directory = commitlog_directory;
-        this.hints_directory = hints_directory;
-        this.cdc_directory = cdc_directory;
-        this.initial_token = initial_token;
-    }
-
-    public static InstanceConfig generate(int nodeNum, File root, String token)
-    {
-        return new InstanceConfig(nodeNum,
-                                  "127.0.0." + nodeNum,
-                                  "127.0.0." + nodeNum,
-                                  "127.0.0." + nodeNum,
-                                  "127.0.0." + nodeNum,
-                                  String.format("%s/node%d/saved_caches", root, nodeNum),
-                                  new String[] { String.format("%s/node%d/data", root, nodeNum) },
-                                  String.format("%s/node%d/commitlog", root, nodeNum),
-                                  String.format("%s/node%d/hints", root, nodeNum),
-                                  String.format("%s/node%d/cdc", root, nodeNum),
-                                  token);
-    }
-}

test/distributed/org/apache/cassandra/distributed/InvokableInstance.java

@@ -1,213 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.distributed;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.Closeable;
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-import java.io.Serializable;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.function.BiConsumer;
-import java.util.function.BiFunction;
-import java.util.function.Consumer;
-import java.util.function.Function;
-
-import org.apache.cassandra.concurrent.NamedThreadFactory;
-import org.apache.cassandra.utils.Throwables;
-
-public abstract class InvokableInstance
-{
-    protected final ExecutorService isolatedExecutor;
-    private final ClassLoader classLoader;
-    private final Method deserializeOnInstance;
-
-    public InvokableInstance(String name, ClassLoader classLoader)
-    {
-        this.isolatedExecutor = Executors.newCachedThreadPool(new NamedThreadFactory(name, Thread.NORM_PRIORITY, classLoader, new ThreadGroup(name)));
-        this.classLoader = classLoader;
-        try
-        {
-            this.deserializeOnInstance = classLoader.loadClass(InvokableInstance.class.getName()).getDeclaredMethod("deserializeOneObject", byte[].class);
-        }
-        catch (ClassNotFoundException | NoSuchMethodException e)
-        {
-            throw new RuntimeException(e);
-        }
-    }
-
-    public interface CallableNoExcept<T> extends Callable<T> { public T call(); }
-    public interface SerializableCallable<T> extends CallableNoExcept<T>, Serializable { }
-    public <T> CallableNoExcept<T> callsOnInstance(SerializableCallable<T> call) { return invokesOnExecutor((SerializableCallable<T>) transferOneObject(call), isolatedExecutor); }
-    public <T> T callOnInstance(SerializableCallable<T> call) { return callsOnInstance(call).call(); }
-
-    public interface SerializableRunnable extends Runnable, Serializable {}
-    public Runnable runsOnInstance(SerializableRunnable run) { return invokesOnExecutor((SerializableRunnable) transferOneObject(run), isolatedExecutor); }
-    public void runOnInstance(SerializableRunnable run) { runsOnInstance(run).run(); }
-
-    public interface SerializableConsumer<T> extends Consumer<T>, Serializable {}
-    public <T> Consumer<T> acceptsOnInstance(SerializableConsumer<T> consumer) { return invokesOnExecutor((SerializableConsumer<T>) transferOneObject(consumer), isolatedExecutor); }
-
-    public interface SerializableBiConsumer<T1, T2> extends BiConsumer<T1, T2>, Serializable {}
-    public <T1, T2> BiConsumer<T1, T2> acceptsOnInstance(SerializableBiConsumer<T1, T2> consumer) { return invokesOnExecutor((SerializableBiConsumer<T1, T2>) transferOneObject(consumer), isolatedExecutor); }
-
-    public interface SerializableFunction<I, O> extends Function<I, O>, Serializable {}
-    public <I, O> Function<I, O> appliesOnInstance(SerializableFunction<I, O> f) { return invokesOnExecutor((SerializableFunction<I, O>) transferOneObject(f), isolatedExecutor); }
-
-    public interface SerializableBiFunction<I1, I2, O> extends BiFunction<I1, I2, O>, Serializable {}
-    public <I1, I2, O> BiFunction<I1, I2, O> appliesOnInstance(SerializableBiFunction<I1, I2, O> f) { return invokesOnExecutor((SerializableBiFunction<I1, I2, O>) transferOneObject(f), isolatedExecutor); }
-
-    public interface TriFunction<I1, I2, I3, O>
-    {
-        O apply(I1 i1, I2 i2, I3 i3);
-    }
-    public interface SerializableTriFunction<I1, I2, I3, O> extends Serializable, TriFunction<I1, I2, I3, O> { }
-
-    public <I1, I2, I3, O> TriFunction<I1, I2, I3, O> appliesOnInstance(SerializableTriFunction<I1, I2, I3, O> f) { return invokesOnExecutor((SerializableTriFunction<I1, I2, I3, O>) transferOneObject(f), isolatedExecutor); }
-
-    public interface InstanceFunction<I, O> extends SerializableBiFunction<Instance, I, O> {}
-
-    // E must be a functional interface, and lambda must be implemented by a lambda function
-    public <E extends Serializable> E invokesOnInstance(E lambda)
-    {
-        return (E) transferOneObject(lambda);
-    }
-
-    public Object transferOneObject(Object object)
-    {
-        byte[] bytes = serializeOneObject(object);
-        try
-        {
-            Object onInstance = deserializeOnInstance.invoke(null, bytes);
-            if (onInstance.getClass().getClassLoader() != classLoader)
-                throw new IllegalStateException(onInstance + " seemingly from wrong class loader: " + onInstance.getClass().getClassLoader() + ", but expected " + classLoader);
-
-            return onInstance;
-        }
-        catch (IllegalAccessException | InvocationTargetException e)
-        {
-            throw new RuntimeException(e);
-        }
-    }
-
-    private byte[] serializeOneObject(Object object)
-    {
-        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
-             ObjectOutputStream oos = new ObjectOutputStream(baos))
-        {
-            oos.writeObject(object);
-            oos.close();
-            return baos.toByteArray();
-        }
-        catch (IOException e)
-        {
-            throw new RuntimeException(e);
-        }
-    }
-
-    @SuppressWarnings("unused") // called through method invocation
-    public static Object deserializeOneObject(byte[] bytes)
-    {
-        try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
-             ObjectInputStream ois = new ObjectInputStream(bais);)
-        {
-            return ois.readObject();
-        }
-        catch (IOException | ClassNotFoundException e)
-        {
-            throw new RuntimeException(e);
-        }
-    }
-
-    private static <V> CallableNoExcept<V> invokesOnExecutor(SerializableCallable<V> callable, ExecutorService invokeOn)
-    {
-        return () -> {
-            try
-            {
-                return invokeOn.submit(callable).get();
-            }
-            catch (InterruptedException e)
-            {
-                throw new RuntimeException(e);
-            }
-            catch (ExecutionException e)
-            {
-                Throwables.maybeFail(e.getCause());
-                throw new AssertionError();
-            }
-        };
-    }
-
-    private static Runnable invokesOnExecutor(SerializableRunnable runnable, ExecutorService invokeOn)
-    {
-        return () -> {
-            try
-            {
-                invokeOn.submit(runnable).get();
-            }
-            catch (InterruptedException e)
-            {
-                throw new RuntimeException(e);
-            }
-            catch (ExecutionException e)
-            {
-                Throwables.maybeFail(e.getCause());
-                throw new AssertionError();
-            }
-        };
-    }
-
-    private static <A> Consumer<A> invokesOnExecutor(SerializableConsumer<A> consumer, ExecutorService invokeOn)
-    {
-        return (a) -> invokesOnExecutor(() -> consumer.accept(a), invokeOn).run();
-    }
-
-    private static <A, B> BiConsumer<A, B> invokesOnExecutor(SerializableBiConsumer<A, B> consumer, ExecutorService invokeOn)
-    {
-        return (a, b) -> invokesOnExecutor(() -> consumer.accept(a, b), invokeOn).run();
-    }
-
-    private static <A, B> Function<A, B> invokesOnExecutor(SerializableFunction<A, B> f, ExecutorService invokeOn)
-    {
-        return (a) -> invokesOnExecutor(() -> f.apply(a), invokeOn).call();
-    }
-
-    private static <A, B, C> BiFunction<A, B, C> invokesOnExecutor(SerializableBiFunction<A, B, C> f, ExecutorService invokeOn)
-    {
-        return (a, b) -> invokesOnExecutor(() -> f.apply(a, b), invokeOn).call();
-    }
-
-    private static <A, B, C, D> SerializableTriFunction<A, B, C, D> invokesOnExecutor(SerializableTriFunction<A, B, C, D> f, ExecutorService invokeOn)
-    {
-        return (a, b, c) -> invokesOnExecutor(() -> f.apply(a, b, c), invokeOn).call();
-    }
-
-    void shutdown()
-    {
-        isolatedExecutor.shutdownNow();
-    }
-
-}

test/distributed/org/apache/cassandra/distributed/TestCluster.java

@@ -1,280 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.distributed;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URLClassLoader;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.LockSupport;
-import java.util.function.IntFunction;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import com.google.common.collect.Sets;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.concurrent.NamedThreadFactory;
-import org.apache.cassandra.db.ColumnFamilyStore;
-import org.apache.cassandra.db.ConsistencyLevel;
-import org.apache.cassandra.db.Keyspace;
-import org.apache.cassandra.io.util.FileUtils;
-import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.net.MessagingService;
-import org.apache.cassandra.utils.FBUtilities;
-
-/**
- * TestCluster creates, initializes and manages Cassandra instances ({@link Instance}.
- *
- * All instances created under the same cluster will have a shared ClassLoader that'll preload
- * common classes required for configuration and communication (byte buffers, primitives, config
- * objects etc). Shared classes are listed in {@link InstanceClassLoader#commonClasses}.
- *
- * Each instance has its own class loader that will load logging, yaml libraries and all non-shared
- * Cassandra package classes. The rule of thumb is that we'd like to have all Cassandra-specific things
- * (unless explitily shared through the common classloader) on a per-classloader basis in order to
- * allow creating more than one instance of DatabaseDescriptor and other Cassandra singletones.
- *
- * All actions (reading, writing, schema changes, etc) are executed by serializing lambda/runnables,
- * transferring them to instance-specific classloaders, deserializing and running them there. Most of
- * the things can be simply captured in closure or passed through `apply` method of the wrapped serializable
- * function/callable. You can use {@link InvokableInstance#{applies|runs|consumes}OnInstance} for executing
- * code on specific instance.
- *
- * Each instance has its own logger. Each instance log line will contain INSTANCE{instance_id}.
- *
- * As of today, messaging is faked by hooking into MessagingService, so we're not using usual Cassandra
- * handlers for internode to have more control over it. Messaging is wired by passing verbs manually.
- * coordinator-handling code and hooks to the callbacks can be found in {@link Coordinator}.
- */
-public class TestCluster implements AutoCloseable
-{
-    // WARNING: we have this logger not (necessarily) for logging, but
-    // to ensure we have instantiated the main classloader's LoggerFactory (and any LogbackStatusListener)
-    // before we instantiate any for a new instance
-    private static final Logger logger = LoggerFactory.getLogger(TestCluster.class);
-
-    private final File root;
-    private final List<Instance> instances;
-    private final Coordinator coordinator;
-    private final Map<InetAddressAndPort, Instance> instanceMap;
-    private final MessageFilters filters;
-
-    private TestCluster(File root, List<Instance> instances)
-    {
-        this.root = root;
-        this.instances = instances;
-        this.instanceMap = new HashMap<>();
-        this.coordinator = new Coordinator(instances.get(0));
-        this.filters = new MessageFilters(this);
-    }
-
-    void launch()
-    {
-        FBUtilities.waitOnFutures(instances.stream()
-                .map(i -> i.isolatedExecutor.submit(() -> i.launch(this)))
-                .collect(Collectors.toList())
-        );
-        for (Instance instance : instances)
-            instanceMap.put(instance.getBroadcastAddress(), instance);
-    }
-
-    public int size()
-    {
-        return instances.size();
-    }
-
-    public Coordinator coordinator()
-    {
-        return coordinator;
-    }
-
-    /**
-     * WARNING: we index from 1 here, for consistency with inet address!
-     */
-    public Instance get(int idx)
-    {
-        return instances.get(idx - 1);
-    }
-
-    public Stream<Instance> stream() { return instances.stream(); }
-
-    public Instance get(InetAddressAndPort addr)
-    {
-        return instanceMap.get(addr);
-    }
-
-    MessageFilters filters()
-    {
-        return filters;
-    }
-
-    MessageFilters.Builder verbs(MessagingService.Verb ... verbs)
-    {
-        return filters.verbs(verbs);
-    }
-
-    public void disableAutoCompaction(String keyspace)
-    {
-        for (Instance instance : instances)
-        {
-            instance.runOnInstance(() -> {
-                for (ColumnFamilyStore cs : Keyspace.open(keyspace).getColumnFamilyStores())
-                    cs.disableAutoCompaction();
-            });
-        }
-    }
-
-    public void schemaChange(String query)
-    {
-        try (SchemaChangeMonitor monitor = new SchemaChangeMonitor())
-        {
-            // execute the schema change
-            coordinator().execute(query, ConsistencyLevel.ALL);
-            monitor.waitForAgreement();
-        }
-    }
-
-    /**
-     * Will wait for a schema change AND agreement that occurs after it is created
-     * (and precedes the invocation to waitForAgreement)
-     *
-     * Works by simply checking if all UUIDs agree after any schema version change event,
-     * so long as the waitForAgreement method has been entered (indicating the change has
-     * taken place on the coordinator)
-     *
-     * This could perhaps be made a little more robust, but this should more than suffice.
-     */
-    public class SchemaChangeMonitor implements AutoCloseable
-    {
-        public SchemaChangeMonitor() {}
-
-        @Override
-        public void close() { }
-
-        public void waitForAgreement()
-        {
-            long start = System.nanoTime();
-            while (1 != instances.stream().map(Instance::getSchemaVersion).distinct().count())
-            {
-                if (System.nanoTime() - start > TimeUnit.MINUTES.toNanos(1L))
-                    throw new IllegalStateException("Schema agreement not reached");
-                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1L));
-            }
-        }
-    }
-
-    public void schemaChange(String statement, int instance)
-    {
-        get(instance).schemaChange(statement);
-    }
-
-    public static TestCluster create(int nodeCount) throws Throwable
-    {
-        return create(nodeCount, Files.createTempDirectory("dtests").toFile());
-    }
-
-    public static TestCluster create(int nodeCount, File root)
-    {
-        root.mkdirs();
-        setupLogging(root);
-
-        IntFunction<ClassLoader> classLoaderFactory = InstanceClassLoader.createFactory(
-                (URLClassLoader) Thread.currentThread().getContextClassLoader());
-        List<Instance> instances = new ArrayList<>();
-        long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount);
-        for (int i = 0 ; i < nodeCount ; ++i)
-        {
-            InstanceConfig instanceConfig = InstanceConfig.generate(i + 1, root, String.valueOf(token));
-            instances.add(new Instance(instanceConfig, classLoaderFactory.apply(i + 1)));
-            token += increment;
-        }
-
-        TestCluster cluster = new TestCluster(root, instances);
-        cluster.launch();
-        return cluster;
-    }
-
-    private static void setupLogging(File root)
-    {
-        try
-        {
-            String testConfPath = "test/conf/logback-dtest.xml";
-            Path logConfPath = Paths.get(root.getPath(), "/logback-dtest.xml");
-            if (!logConfPath.toFile().exists())
-            {
-                Files.copy(new File(testConfPath).toPath(),
-                           logConfPath);
-            }
-            System.setProperty("logback.configurationFile", "file://" + logConfPath);
-        }
-        catch (IOException e)
-        {
-            throw new RuntimeException(e);
-        }
-    }
-
-    @Override
-    public void close()
-    {
-        List<Future<?>> futures = instances.stream()
-                .map(i -> i.isolatedExecutor.submit(i::shutdown))
-                .collect(Collectors.toList());
-
-        // Make sure to only delete directory when threads are stopped
-        FBUtilities.waitOnFutures(futures, 60, TimeUnit.SECONDS);
-        FileUtils.deleteRecursive(root);
-
-        //withThreadLeakCheck(futures);
-        System.gc();
-    }
-
-    // We do not want this check to run every time until we fix problems with tread stops
-    private void withThreadLeakCheck(List<Future<?>> futures)
-    {
-        FBUtilities.waitOnFutures(futures);
-
-        Set<Thread> threadSet = Thread.getAllStackTraces().keySet();
-        threadSet = Sets.difference(threadSet, Collections.singletonMap(Thread.currentThread(), null).keySet());
-        if (!threadSet.isEmpty())
-        {
-            for (Thread thread : threadSet)
-            {
-                System.out.println(thread);
-                System.out.println(Arrays.toString(thread.getStackTrace()));
-            }
-            throw new RuntimeException(String.format("Not all threads have shut down. %d threads are still running: %s", threadSet.size(), threadSet));
-        }
-    }
-
-}
-

test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.List;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.IUpgradeableInstance;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.distributed.impl.Versions;
+
+/**
+ * A multi-version cluster, offering only the cross-version API
+ *
+ * TODO: we could perhaps offer some convenience methods for nodes we know to be of the 'current' version,
+ * to permit upgrade tests to perform cluster operations without updating the cross-version API,
+ * so long as one node is up-to-date.
+ */
+public class UpgradeableCluster extends AbstractCluster<IUpgradeableInstance> implements ICluster, AutoCloseable
+{
+    private UpgradeableCluster(File root, Versions.Version version, List<InstanceConfig> configs, ClassLoader sharedClassLoader)
+    {
+        super(root, version, configs, sharedClassLoader);
+    }
+
+    protected IUpgradeableInstance newInstanceWrapper(Versions.Version version, InstanceConfig config)
+    {
+        return new Wrapper(version, config);
+    }
+
+    public static UpgradeableCluster create(int nodeCount) throws Throwable
+    {
+        return create(nodeCount, UpgradeableCluster::new);
+    }
+    public static UpgradeableCluster create(int nodeCount, File root)
+    {
+        return create(nodeCount, Versions.CURRENT, root, UpgradeableCluster::new);
+    }
+
+    public static UpgradeableCluster create(int nodeCount, Versions.Version version) throws IOException
+    {
+        return create(nodeCount, version, Files.createTempDirectory("dtests").toFile(), UpgradeableCluster::new);
+    }
+    public static UpgradeableCluster create(int nodeCount, Versions.Version version, File root)
+    {
+        return create(nodeCount, version, root, UpgradeableCluster::new);
+    }
+
+}
+

test/distributed/org/apache/cassandra/distributed/api/ICluster.java

@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+import java.util.stream.Stream;
+
+public interface ICluster
+{
+
+    IInstance get(int i);
+    IInstance get(InetAddressAndPort endpoint);
+    int size();
+    Stream<? extends IInstance> stream();
+    IMessageFilters filters();
+
+}

test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+// The cross-version API requires that a Coordinator can be constructed without any constructor arguments
+public interface ICoordinator
+{
+    // a bit hacky, but ConsistencyLevel draws in too many dependent classes, so we cannot have a cross-version
+    // method signature that accepts ConsistencyLevel directly.  So we just accept an Enum<?> and cast.
+    Object[][] execute(String query, Enum<?> consistencyLevel, Object... boundValues);
+}

test/distributed/org/apache/cassandra/distributed/api/IInstance.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+import java.util.UUID;
+
+// The cross-version API requires that an Instance has a constructor signature of (IInstanceConfig, ClassLoader)
+public interface IInstance extends IIsolatedExecutor
+{
+    ICoordinator coordinator();
+    IListen listen();
+
+    void schemaChangeInternal(String query);
+    public Object[][] executeInternal(String query, Object... args);
+
+    IInstanceConfig config();
+    public InetAddressAndPort broadcastAddressAndPort();
+    UUID schemaVersion();
+
+    void startup();
+    void shutdown();
+
+    // these methods are not for external use, but for simplicity we leave them public and on the normal IInstance interface
+    void startup(ICluster cluster);
+    void receiveMessage(IMessage message);
+}

test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+import java.util.UUID;
+
+public interface IInstanceConfig
+{
+    int num();
+    UUID hostId();
+    InetAddressAndPort broadcastAddressAndPort();
+
+    /**
+     * write the specified parameters to the Config object; we do not specify Config as the type to support a Config
+     * from any ClassLoader; the implementation must not directly access any fields of the Object, or cast it, but
+     * must use the reflection API to modify the state
+     */
+    void propagate(Object writeToConfig);
+
+    Object get(String fieldName);
+    String getString(String fieldName);
+    int getInt(String fieldName);
+}

test/distributed/org/apache/cassandra/distributed/api/IIsolatedExecutor.java

@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+import java.io.Serializable;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+/**
+ * Represents a clean way to handoff evaluation of some work to an executor associated
+ * with a node's lifetime.
+ *
+ * There is no transfer of execution to the parallel class hierarchy.
+ *
+ * Classes, such as Instance, that are themselves instantiated on the correct ClassLoader, utilise this class
+ * to ensure the lifetime of any thread evaluating one of its method invocations matches the lifetime of the class itself.
+ * Since they are instantiated on the correct ClassLoader, sharing only the interface, there is no serialization necessary.
+ */
+public interface IIsolatedExecutor
+{
+    public interface CallableNoExcept<O> extends Callable<O> { public O call(); }
+    public interface SerializableCallable<O> extends CallableNoExcept<O>, Serializable { }
+    public interface SerializableRunnable extends Runnable, Serializable {}
+    public interface SerializableConsumer<O> extends Consumer<O>, Serializable {}
+    public interface SerializableBiConsumer<I1, I2> extends BiConsumer<I1, I2>, Serializable {}
+    public interface SerializableFunction<I, O> extends Function<I, O>, Serializable {}
+    public interface SerializableBiFunction<I1, I2, O> extends BiFunction<I1, I2, O>, Serializable {}
+    public interface TriFunction<I1, I2, I3, O>
+    {
+        O apply(I1 i1, I2 i2, I3 i3);
+    }
+    public interface SerializableTriFunction<I1, I2, I3, O> extends Serializable, TriFunction<I1, I2, I3, O> { }
+
+    void shutdown();
+
+    /**
+     * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
+     */
+    <O> CallableNoExcept<Future<O>> async(CallableNoExcept<O> call);
+
+    /**
+     * Convert the execution to one performed synchronously on the IsolatedExecutor
+     */
+    <O> CallableNoExcept<O> sync(CallableNoExcept<O> call);
+
+    /**
+     * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
+     */
+    CallableNoExcept<Future<?>> async(Runnable run);
+
+    /**
+     * Convert the execution to one performed synchronously on the IsolatedExecutor
+     */
+    Runnable sync(Runnable run);
+
+    /**
+     * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
+     */
+    <I> Function<I, Future<?>> async(Consumer<I> consumer);
+
+    /**
+     * Convert the execution to one performed synchronously on the IsolatedExecutor
+     */
+    <I> Consumer<I> sync(Consumer<I> consumer);
+
+    /**
+     * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
+     */
+    <I1, I2> BiFunction<I1, I2, Future<?>> async(BiConsumer<I1, I2> consumer);
+
+    /**
+     * Convert the execution to one performed synchronously on the IsolatedExecutor
+     */
+    <I1, I2> BiConsumer<I1, I2> sync(BiConsumer<I1, I2> consumer);
+
+    /**
+     * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
+     */
+    <I, O> Function<I, Future<O>> async(Function<I, O> f);
+
+    /**
+     * Convert the execution to one performed synchronously on the IsolatedExecutor
+     */
+    <I, O> Function<I, O> sync(Function<I, O> f);
+
+    /**
+     * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
+     */
+    <I1, I2, O> BiFunction<I1, I2, Future<O>> async(BiFunction<I1, I2, O> f);
+
+    /**
+     * Convert the execution to one performed synchronously on the IsolatedExecutor
+     */
+    <I1, I2, O> BiFunction<I1, I2, O> sync(BiFunction<I1, I2, O> f);
+
+    /**
+     * Convert the execution to one performed asynchronously on the IsolatedExecutor, returning a Future of the execution result
+     */
+    <I1, I2, I3, O> TriFunction<I1, I2, I3, Future<O>> async(TriFunction<I1, I2, I3, O> f);
+
+    /**
+     * Convert the execution to one performed synchronously on the IsolatedExecutor
+     */
+    <I1, I2, I3, O> TriFunction<I1, I2, I3, O> sync(TriFunction<I1, I2, I3, O> f);
+
+}

test/distributed/org/apache/cassandra/distributed/api/IListen.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+public interface IListen
+{
+    public interface Cancel { void cancel(); }
+
+    Cancel schema(Runnable onChange);
+}

test/distributed/org/apache/cassandra/distributed/api/IMessage.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+/**
+ * A cross-version interface for delivering internode messages via message sinks
+ */
+public interface IMessage
+{
+    int verb();
+    byte[] bytes();
+    int id();
+    int version();
+    InetAddressAndPort from();
+}

test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java

@@ -16,27 +16,33 @@
  * limitations under the License.
  */
 
-package org.apache.cassandra.distributed;
+package org.apache.cassandra.distributed.api;
 
-import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.net.MessagingService;
 
-import java.net.InetAddress;
+import java.util.function.BiConsumer;
 
-public class LegacyAdapter
+public interface IMessageFilters
 {
-    private static final InetAddressAndPort broadcastAddressAndPort;
+    public interface Filter
-    static
     {
-        InetAddress address = FBUtilities.getBroadcastAddress();
+        Filter restore();
-        int port = DatabaseDescriptor.getStoragePort();
+        Filter drop();
-        broadcastAddressAndPort = InetAddressAndPort.getByAddressOverrideDefaults(address, port);
     }
 
-    public static InetAddressAndPort getBroadcastAddressAndPort()
+    public interface Builder
     {
-        return broadcastAddressAndPort;
+        Builder from(int ... nums);
+        Builder to(int ... nums);
+        Filter ready();
+        Filter drop();
     }
 
+    Builder verbs(MessagingService.Verb... verbs);
+    Builder allVerbs();
+    void reset();
+
+    // internal
+    BiConsumer<InetAddressAndPort, IMessage> filter(BiConsumer<InetAddressAndPort, IMessage> applyIfNotFiltered);
 }

test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java

@@ -0,0 +1,406 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.collect.Sets;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.distributed.api.ICoordinator;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.api.IListen;
+import org.apache.cassandra.distributed.api.IMessage;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.concurrent.SimpleCondition;
+
+/**
+ * AbstractCluster creates, initializes and manages Cassandra instances ({@link Instance}.
+ *
+ * All instances created under the same cluster will have a shared ClassLoader that'll preload
+ * common classes required for configuration and communication (byte buffers, primitives, config
+ * objects etc). Shared classes are listed in {@link InstanceClassLoader}.
+ *
+ * Each instance has its own class loader that will load logging, yaml libraries and all non-shared
+ * Cassandra package classes. The rule of thumb is that we'd like to have all Cassandra-specific things
+ * (unless explitily shared through the common classloader) on a per-classloader basis in order to
+ * allow creating more than one instance of DatabaseDescriptor and other Cassandra singletones.
+ *
+ * All actions (reading, writing, schema changes, etc) are executed by serializing lambda/runnables,
+ * transferring them to instance-specific classloaders, deserializing and running them there. Most of
+ * the things can be simply captured in closure or passed through `apply` method of the wrapped serializable
+ * function/callable. You can use {@link Instance#{applies|runs|consumes}OnInstance} for executing
+ * code on specific instance.
+ *
+ * Each instance has its own logger. Each instance log line will contain INSTANCE{instance_id}.
+ *
+ * As of today, messaging is faked by hooking into MessagingService, so we're not using usual Cassandra
+ * handlers for internode to have more control over it. Messaging is wired by passing verbs manually.
+ * coordinator-handling code and hooks to the callbacks can be found in {@link Coordinator}.
+ */
+public abstract class AbstractCluster<I extends IInstance> implements ICluster, AutoCloseable
+{
+    // WARNING: we have this logger not (necessarily) for logging, but
+    // to ensure we have instantiated the main classloader's LoggerFactory (and any LogbackStatusListener)
+    // before we instantiate any for a new instance
+    private static final Logger logger = LoggerFactory.getLogger(AbstractCluster.class);
+
+    private final File root;
+    private final ClassLoader sharedClassLoader;
+
+    // mutated by starting/stopping a node
+    private final List<I> instances;
+    private final Map<InetAddressAndPort, I> instanceMap;
+
+    // mutated by user-facing API
+    private final MessageFilters filters;
+
+    protected class Wrapper extends DelegatingInvokableInstance implements IUpgradeableInstance
+    {
+        private final InstanceConfig config;
+        private volatile IInvokableInstance delegate;
+        private volatile Versions.Version version;
+        private volatile boolean isShutdown = true;
+
+        protected IInvokableInstance delegate()
+        {
+            if (delegate == null)
+                delegate = newInstance();
+            return delegate;
+        }
+
+        public Wrapper(Versions.Version version, InstanceConfig config)
+        {
+            this.config = config;
+            this.version = version;
+            // we ensure there is always a non-null delegate, so that the executor may be used while the node is offline
+            this.delegate = newInstance();
+        }
+
+        private IInvokableInstance newInstance()
+        {
+            ClassLoader classLoader = new InstanceClassLoader(config.num(), version.classpath, sharedClassLoader);
+            return Instance.transferAdhoc((SerializableBiFunction<IInstanceConfig, ClassLoader, Instance>)Instance::new, classLoader)
+                           .apply(config, classLoader);
+        }
+
+        public IInstanceConfig config()
+        {
+            return config;
+        }
+
+        @Override
+        public synchronized void startup()
+        {
+            if (!isShutdown)
+                throw new IllegalStateException();
+            delegate().startup(AbstractCluster.this);
+            isShutdown = false;
+        }
+
+        @Override
+        public synchronized void shutdown()
+        {
+            if (isShutdown)
+                throw new IllegalStateException();
+            isShutdown = true;
+            delegate.shutdown();
+            delegate = null;
+        }
+
+        @Override
+        public void receiveMessage(IMessage message)
+        {
+            IInvokableInstance delegate = this.delegate;
+            if (!isShutdown && delegate != null) // since we sync directly on the other node, we drop messages immediately if we are shutdown
+                delegate.receiveMessage(message);
+        }
+
+        @Override
+        public synchronized void setVersion(Versions.Version version)
+        {
+            if (!isShutdown)
+                throw new IllegalStateException("Must be shutdown before version can be modified");
+            // re-initialise
+            this.version = version;
+            if (delegate != null)
+            {
+                // we can have a non-null delegate even thought we are shutdown, if delegate() has been invoked since shutdown.
+                delegate.shutdown();
+                delegate = null;
+            }
+        }
+    }
+
+    protected AbstractCluster(File root, Versions.Version version, List<InstanceConfig> configs, ClassLoader sharedClassLoader)
+    {
+        this.root = root;
+        this.sharedClassLoader = sharedClassLoader;
+        this.instances = new ArrayList<>();
+        this.instanceMap = new HashMap<>();
+        for (InstanceConfig config : configs)
+        {
+            I instance = newInstanceWrapper(version, config);
+            instances.add(instance);
+            // we use the config().broadcastAddressAndPort() here because we have not initialised the Instance
+            I prev = instanceMap.put(instance.broadcastAddressAndPort(), instance);
+            if (null != prev)
+                throw new IllegalStateException("Cluster cannot have multiple nodes with same InetAddressAndPort: " + instance.broadcastAddressAndPort() + " vs " + prev.broadcastAddressAndPort());
+        }
+        this.filters = new MessageFilters(this);
+    }
+
+    protected abstract I newInstanceWrapper(Versions.Version version, InstanceConfig config);
+
+    /**
+     * WARNING: we index from 1 here, for consistency with inet address!
+     */
+    public ICoordinator coordinator(int node)
+    {
+        return instances.get(node - 1).coordinator();
+    }
+    /**
+     * WARNING: we index from 1 here, for consistency with inet address!
+     */
+    public I get(int node) { return instances.get(node - 1); }
+    public I get(InetAddressAndPort addr) { return instanceMap.get(addr); }
+
+    public int size()
+    {
+        return instances.size();
+    }
+    public Stream<I> stream() { return instances.stream(); }
+    public void forEach(IIsolatedExecutor.SerializableRunnable runnable) { forEach(i -> i.sync(runnable)); }
+    public void forEach(Consumer<? super I> consumer) { instances.forEach(consumer); }
+    public void parallelForEach(IIsolatedExecutor.SerializableConsumer<? super I> consumer, long timeout, TimeUnit units)
+    {
+        FBUtilities.waitOnFutures(instances.stream()
+                                           .map(i -> i.async(consumer).apply(i))
+                                           .collect(Collectors.toList()),
+                                  timeout, units);
+    }
+
+
+    public IMessageFilters filters() { return filters; }
+    public MessageFilters.Builder verbs(MessagingService.Verb ... verbs) { return filters.verbs(verbs); }
+
+    public void disableAutoCompaction(String keyspace)
+    {
+        forEach(() -> {
+            for (ColumnFamilyStore cs : Keyspace.open(keyspace).getColumnFamilyStores())
+                cs.disableAutoCompaction();
+        });
+    }
+
+    public void schemaChange(String query)
+    {
+        try (SchemaChangeMonitor monitor = new SchemaChangeMonitor())
+        {
+            // execute the schema change
+            coordinator(1).execute(query, ConsistencyLevel.ALL);
+            monitor.waitForAgreement();
+        }
+    }
+
+    /**
+     * Will wait for a schema change AND agreement that occurs after it is created
+     * (and precedes the invocation to waitForAgreement)
+     *
+     * Works by simply checking if all UUIDs agree after any schema version change event,
+     * so long as the waitForAgreement method has been entered (indicating the change has
+     * taken place on the coordinator)
+     *
+     * This could perhaps be made a little more robust, but this should more than suffice.
+     */
+    public class SchemaChangeMonitor implements AutoCloseable
+    {
+        final List<IListen.Cancel> cleanup;
+        volatile boolean schemaHasChanged;
+        final SimpleCondition agreement = new SimpleCondition();
+
+        public SchemaChangeMonitor()
+        {
+            this.cleanup = new ArrayList<>(instances.size());
+            for (IInstance instance : instances)
+                cleanup.add(instance.listen().schema(this::signal));
+        }
+
+        private void signal()
+        {
+            if (schemaHasChanged && 1 == instances.stream().map(IInstance::schemaVersion).distinct().count())
+                agreement.signalAll();
+        }
+
+        @Override
+        public void close()
+        {
+            for (IListen.Cancel cancel : cleanup)
+                cancel.cancel();
+        }
+
+        public void waitForAgreement()
+        {
+            schemaHasChanged = true;
+            signal();
+            try
+            {
+                if (!agreement.await(1L, TimeUnit.MINUTES))
+                    throw new InterruptedException();
+            }
+            catch (InterruptedException e)
+            {
+                throw new IllegalStateException("Schema agreement not reached");
+            }
+        }
+    }
+
+    public void schemaChange(String statement, int instance)
+    {
+        get(instance).schemaChangeInternal(statement);
+    }
+
+    void startup()
+    {
+        parallelForEach(I::startup, 0, null);
+    }
+
+    protected interface Factory<I extends IInstance, C extends AbstractCluster<I>>
+    {
+        C newCluster(File root, Versions.Version version, List<InstanceConfig> configs, ClassLoader sharedClassLoader);
+    }
+
+    protected static <I extends IInstance, C extends AbstractCluster<I>> C
+    create(int nodeCount, Factory<I, C> factory) throws Throwable
+    {
+        return create(nodeCount, Files.createTempDirectory("dtests").toFile(), factory);
+    }
+
+    protected static <I extends IInstance, C extends AbstractCluster<I>> C
+    create(int nodeCount, File root, Factory<I, C> factory)
+    {
+        return create(nodeCount, Versions.CURRENT, root, factory);
+    }
+
+    protected static <I extends IInstance, C extends AbstractCluster<I>> C
+    create(int nodeCount, Versions.Version version, Factory<I, C> factory) throws IOException
+    {
+        return create(nodeCount, version, Files.createTempDirectory("dtests").toFile(), factory);
+    }
+
+    protected static <I extends IInstance, C extends AbstractCluster<I>> C
+    create(int nodeCount, Versions.Version version, File root, Factory<I, C> factory)
+    {
+        root.mkdirs();
+        setupLogging(root);
+
+        ClassLoader sharedClassLoader = Thread.currentThread().getContextClassLoader();
+
+        List<InstanceConfig> configs = new ArrayList<>();
+        long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount);
+        for (int i = 0 ; i < nodeCount ; ++i)
+        {
+            InstanceConfig config = InstanceConfig.generate(i + 1, root, String.valueOf(token));
+            configs.add(config);
+            token += increment;
+        }
+
+        C cluster = factory.newCluster(root, version, configs, sharedClassLoader);
+        cluster.startup();
+        return cluster;
+    }
+
+    private static void setupLogging(File root)
+    {
+        try
+        {
+            String testConfPath = "test/conf/logback-dtest.xml";
+            Path logConfPath = Paths.get(root.getPath(), "/logback-dtest.xml");
+            if (!logConfPath.toFile().exists())
+            {
+                Files.copy(new File(testConfPath).toPath(),
+                           logConfPath);
+            }
+            System.setProperty("logback.configurationFile", "file://" + logConfPath);
+        }
+        catch (IOException e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public void close()
+    {
+        parallelForEach(IInstance::shutdown, 1L, TimeUnit.MINUTES);
+
+        // Make sure to only delete directory when threads are stopped
+        FileUtils.deleteRecursive(root);
+
+        //withThreadLeakCheck(futures);
+        System.gc();
+    }
+
+    // We do not want this check to run every time until we fix problems with tread stops
+    private void withThreadLeakCheck(List<Future<?>> futures)
+    {
+        FBUtilities.waitOnFutures(futures);
+
+        Set<Thread> threadSet = Thread.getAllStackTraces().keySet();
+        threadSet = Sets.difference(threadSet, Collections.singletonMap(Thread.currentThread(), null).keySet());
+        if (!threadSet.isEmpty())
+        {
+            for (Thread thread : threadSet)
+            {
+                System.out.println(thread);
+                System.out.println(Arrays.toString(thread.getStackTrace()));
+            }
+            throw new RuntimeException(String.format("Not all threads have shut down. %d threads are still running: %s", threadSet.size(), threadSet));
+        }
+    }
+
+}
+

test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java

@@ -16,8 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.cassandra.distributed;
+package org.apache.cassandra.distributed.impl;
-
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -27,53 +26,51 @@ import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.cql3.QueryProcessor;
 import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.ICoordinator;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.QueryState;
 import org.apache.cassandra.transport.ProtocolVersion;
 import org.apache.cassandra.transport.messages.ResultMessage;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
-
+public class Coordinator implements ICoordinator
-public class Coordinator
 {
     final Instance instance;
-
     public Coordinator(Instance instance)
     {
         this.instance = instance;
     }
 
-    private static Object[][] coordinatorExecute(String query, int consistencyLevel, Object[] bindings)
+    @Override
+    public Object[][] execute(String query, Enum<?> consistencyLevelOrigin, Object... boundValues)
     {
-        CQLStatement prepared = QueryProcessor.getStatement(query, ClientState.forInternalCalls()).statement;
+        return instance.sync(() -> {
-        List<ByteBuffer> boundValues = new ArrayList<>();
+            ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name());
-        for (Object binding : bindings)
+            CQLStatement prepared = QueryProcessor.getStatement(query, ClientState.forInternalCalls()).statement;
-        {
+            List<ByteBuffer> boundBBValues = new ArrayList<>();
-            boundValues.add(ByteBufferUtil.objectToBytes(binding));
+            for (Object boundValue : boundValues)
-        }
+            {
+                boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue));
+            }
 
-        ResultMessage res = prepared.execute(QueryState.forInternalCalls(),
+            ResultMessage res = prepared.execute(QueryState.forInternalCalls(),
-                                             QueryOptions.create(ConsistencyLevel.fromCode(consistencyLevel),
+                                                 QueryOptions.create(consistencyLevel,
-                                                                 boundValues,
+                                                                     boundBBValues,
-                                                                 false,
+                                                                     false,
-                                                                 10,
+                                                                     10,
-                                                                 null,
+                                                                     null,
-                                                                 null,
+                                                                     null,
-                                                                 ProtocolVersion.V4),
+                                                                     ProtocolVersion.CURRENT),
-                                              System.nanoTime());
+                                                  System.nanoTime());
 
-        if (res != null && res.kind == ResultMessage.Kind.ROWS)
+            if (res != null && res.kind == ResultMessage.Kind.ROWS)
-        {
+            {
-            return RowUtil.toObjects((ResultMessage.Rows) res);
+                return RowUtil.toObjects((ResultMessage.Rows) res);
-        }
+            }
-        else
+            else
-        {
+            {
-            return new Object[][]{};
+                return new Object[][]{};
-        }
+            }
-    }
+        }).call();
-
-    public Object[][] execute(String query, ConsistencyLevel consistencyLevel, Object... boundValues)
-    {
-        return instance.appliesOnInstance(Coordinator::coordinatorExecute).apply(query, consistencyLevel.code, boundValues);
     }
 }

test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java

@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import java.io.Serializable;
+import java.util.UUID;
+import java.util.concurrent.Future;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.ICoordinator;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IListen;
+import org.apache.cassandra.distributed.api.IMessage;
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+public abstract class DelegatingInvokableInstance implements IInvokableInstance
+{
+    protected abstract IInvokableInstance delegate();
+    
+    @Override
+    public <E extends Serializable> E transfer(E object)
+    {
+        return delegate().transfer(object);
+    }
+
+    @Override
+    public InetAddressAndPort broadcastAddressAndPort()
+    {
+        return delegate().broadcastAddressAndPort();
+    }
+
+    @Override
+    public Object[][] executeInternal(String query, Object... args)
+    {
+        return delegate().executeInternal(query, args);
+    }
+
+    @Override
+    public UUID schemaVersion()
+    {
+        return delegate().schemaVersion();
+    }
+
+    @Override
+    public void startup()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void schemaChangeInternal(String query)
+    {
+        delegate().schemaChangeInternal(query);
+    }
+
+    @Override
+    public IInstanceConfig config()
+    {
+        return delegate().config();
+    }
+
+    @Override
+    public ICoordinator coordinator()
+    {
+        // TODO: stash and clear coordinator on startup/shutdown?
+        return delegate().coordinator();
+    }
+
+    public IListen listen()
+    {
+        return delegate().listen();
+    }
+
+    @Override
+    public void shutdown()
+    {
+        delegate().shutdown();
+    }
+
+    @Override
+    public void startup(ICluster cluster)
+    {
+        delegate().startup(cluster);
+    }
+
+    @Override
+    public void receiveMessage(IMessage message)
+    {
+        delegate().receiveMessage(message);
+    }
+
+    @Override
+    public <O> CallableNoExcept<Future<O>> async(CallableNoExcept<O> call)
+    {
+        return delegate().async(call);
+    }
+
+    @Override
+    public <O> CallableNoExcept<O> sync(CallableNoExcept<O> call)
+    {
+        return delegate().sync(call);
+    }
+
+    @Override
+    public CallableNoExcept<Future<?>> async(Runnable run)
+    {
+        return delegate().async(run);
+    }
+
+    @Override
+    public Runnable sync(Runnable run)
+    {
+        return delegate().sync(run);
+    }
+
+    @Override
+    public <I> Function<I, Future<?>> async(Consumer<I> consumer)
+    {
+        return delegate().async(consumer);
+    }
+
+    @Override
+    public <I> Consumer<I> sync(Consumer<I> consumer)
+    {
+        return delegate().sync(consumer);
+    }
+
+    @Override
+    public <I1, I2> BiFunction<I1, I2, Future<?>> async(BiConsumer<I1, I2> consumer)
+    {
+        return delegate().async(consumer);
+    }
+
+    @Override
+    public <I1, I2> BiConsumer<I1, I2> sync(BiConsumer<I1, I2> consumer)
+    {
+        return delegate().sync(consumer);
+    }
+
+    @Override
+    public <I, O> Function<I, Future<O>> async(Function<I, O> f)
+    {
+        return delegate().async(f);
+    }
+
+    @Override
+    public <I, O> Function<I, O> sync(Function<I, O> f)
+    {
+        return delegate().sync(f);
+    }
+
+    @Override
+    public <I1, I2, O> BiFunction<I1, I2, Future<O>> async(BiFunction<I1, I2, O> f)
+    {
+        return delegate().async(f);
+    }
+
+    @Override
+    public <I1, I2, O> BiFunction<I1, I2, O> sync(BiFunction<I1, I2, O> f)
+    {
+        return delegate().sync(f);
+    }
+
+    @Override
+    public <I1, I2, I3, O> TriFunction<I1, I2, I3, Future<O>> async(TriFunction<I1, I2, I3, O> f)
+    {
+        return delegate().async(f);
+    }
+
+    @Override
+    public <I1, I2, I3, O> TriFunction<I1, I2, I3, O> sync(TriFunction<I1, I2, I3, O> f)
+    {
+        return delegate().sync(f);
+    }
+
+}

test/distributed/org/apache/cassandra/distributed/impl/IInvokableInstance.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import java.io.Serializable;
+import java.util.concurrent.Future;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+
+/**
+ * This version is only supported for a Cluster running the same code as the test environment, and permits
+ * ergonomic cross-node behaviours, without editing the cross-version API.
+ *
+ * A lambda can be written tto be invoked on any or all of the nodes.
+ *
+ * The reason this cannot (easily) be made cross-version is that the lambda is tied to the declaring class, which will
+ * not be the same in the alternate version.  Even were it not, there would likely be a runtime linkage error given
+ * any code divergence.
+ */
+public interface IInvokableInstance extends IInstance
+{
+    public default <O> CallableNoExcept<Future<O>> asyncCallsOnInstance(SerializableCallable<O> call) { return async(transfer(call)); }
+    public default <O> CallableNoExcept<O> callsOnInstance(SerializableCallable<O> call) { return sync(transfer(call)); }
+    public default <O> O callOnInstance(SerializableCallable<O> call) { return callsOnInstance(call).call(); }
+
+    public default CallableNoExcept<Future<?>> asyncRunsOnInstance(SerializableRunnable run) { return async(transfer(run)); }
+    public default Runnable runsOnInstance(SerializableRunnable run) { return sync(transfer(run)); }
+    public default void runOnInstance(SerializableRunnable run) { runsOnInstance(run).run(); }
+
+    public default <I> Function<I, Future<?>> asyncAcceptsOnInstance(SerializableConsumer<I> consumer) { return async(transfer(consumer)); }
+    public default <I> Consumer<I> acceptsOnInstance(SerializableConsumer<I> consumer) { return sync(transfer(consumer)); }
+
+    public default <I1, I2> BiFunction<I1, I2, Future<?>> asyncAcceptsOnInstance(SerializableBiConsumer<I1, I2> consumer) { return async(transfer(consumer)); }
+    public default <I1, I2> BiConsumer<I1, I2> acceptsOnInstance(SerializableBiConsumer<I1, I2> consumer) { return sync(transfer(consumer)); }
+
+    public default <I, O> Function<I, Future<O>> asyncAppliesOnInstance(SerializableFunction<I, O> f) { return async(transfer(f)); }
+    public default <I, O> Function<I, O> appliesOnInstance(SerializableFunction<I, O> f) { return sync(transfer(f)); }
+
+    public default <I1, I2, O> BiFunction<I1, I2, Future<O>> asyncAppliesOnInstance(SerializableBiFunction<I1, I2, O> f) { return async(transfer(f)); }
+    public default <I1, I2, O> BiFunction<I1, I2, O> appliesOnInstance(SerializableBiFunction<I1, I2, O> f) { return sync(transfer(f)); }
+
+    public default <I1, I2, I3, O> TriFunction<I1, I2, I3, Future<O>> asyncAppliesOnInstance(SerializableTriFunction<I1, I2, I3, O> f) { return async(transfer(f)); }
+    public default <I1, I2, I3, O> TriFunction<I1, I2, I3, O> appliesOnInstance(SerializableTriFunction<I1, I2, I3, O> f) { return sync(transfer(f)); }
+
+    public <E extends Serializable> E transfer(E object);
+
+}

test/distributed/org/apache/cassandra/distributed/impl/IUpgradeableInstance.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import org.apache.cassandra.distributed.api.IInstance;
+
+// this lives outside the api package so that we do not have to worry about inter-version compatibility
+public interface IUpgradeableInstance extends IInstance
+{
+    // only to be invoked while the node is shutdown!
+    public void setVersion(Versions.Version version);
+}

test/distributed/org/apache/cassandra/distributed/impl/Instance.java

@@ -16,12 +16,11 @@
  * limitations under the License.
  */
 
-package org.apache.cassandra.distributed;
+package org.apache.cassandra.distributed.impl;
 
 import java.io.File;
 import java.io.IOException;
 import java.net.InetAddress;
-import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -42,7 +41,6 @@ import org.apache.cassandra.concurrent.SharedExecutorPool;
 import org.apache.cassandra.concurrent.StageManager;
 import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.config.Schema;
 import org.apache.cassandra.config.SchemaConstants;
 import org.apache.cassandra.cql3.CQLStatement;
@@ -57,6 +55,11 @@ import org.apache.cassandra.db.commitlog.CommitLog;
 import org.apache.cassandra.db.compaction.CompactionManager;
 import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.api.ICoordinator;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IListen;
+import org.apache.cassandra.distributed.api.IMessage;
+import org.apache.cassandra.distributed.api.ICluster;
 import org.apache.cassandra.gms.ApplicationState;
 import org.apache.cassandra.gms.Gossiper;
 import org.apache.cassandra.gms.VersionedValue;
@@ -66,13 +69,12 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.DataInputBuffer;
 import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.locator.SimpleSeedProvider;
-import org.apache.cassandra.locator.SimpleSnitch;
 import org.apache.cassandra.net.IMessageSink;
 import org.apache.cassandra.net.MessageDeliveryTask;
 import org.apache.cassandra.net.MessageIn;
 import org.apache.cassandra.net.MessageOut;
 import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.LegacySchemaMigrator;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.PendingRangeCalculatorService;
 import org.apache.cassandra.service.QueryState;
@@ -83,44 +85,69 @@ import org.apache.cassandra.utils.Throwables;
 import org.apache.cassandra.utils.concurrent.Ref;
 import org.apache.cassandra.utils.memory.BufferPool;
 
-public class Instance extends InvokableInstance
+public class Instance extends IsolatedExecutor implements IInvokableInstance
 {
-    public final InstanceConfig config;
+    public final IInstanceConfig config;
 
-    public Instance(InstanceConfig config, ClassLoader classLoader)
+    // should never be invoked directly, so that it is instantiated on other class loader;
+    // only visible for inheritance
+    Instance(IInstanceConfig config, ClassLoader classLoader)
     {
-        super("node" + config.num, classLoader);
+        super("node" + config.num(), classLoader);
         this.config = config;
+        InstanceIDDefiner.setInstanceId(config.num());
+        FBUtilities.setBroadcastInetAddress(config.broadcastAddressAndPort().address);
     }
 
-    public InetAddressAndPort getBroadcastAddress() { return callOnInstance(LegacyAdapter::getBroadcastAddressAndPort); }
+    public IInstanceConfig config()
+    {
+        return config;
+    }
+
+    public ICoordinator coordinator()
+    {
+        return new Coordinator(this);
+    }
+
+    public IListen listen()
+    {
+        return new Listen(this);
+    }
+
+    @Override
+    public InetAddressAndPort broadcastAddressAndPort() { return config.broadcastAddressAndPort(); }
 
     public Object[][] executeInternal(String query, Object... args)
     {
-        return callOnInstance(() ->
+        return sync(() -> {
-        {
             ParsedStatement.Prepared prepared = QueryProcessor.prepareInternal(query);
             ResultMessage result = prepared.statement.executeInternal(QueryProcessor.internalQueryState(),
-                    QueryProcessor.makeInternalOptions(prepared, args));
+                                                                      QueryProcessor.makeInternalOptions(prepared, args));
 
             if (result instanceof ResultMessage.Rows)
                 return RowUtil.toObjects((ResultMessage.Rows)result);
             else
                 return null;
-        });
+        }).call();
     }
 
-    public UUID getSchemaVersion()
+    @Override
+    public UUID schemaVersion()
     {
-        // we do not use method reference syntax here, because we need to invoke on the node-local schema instance
+        // we do not use method reference syntax here, because we need to sync on the node-local schema instance
         //noinspection Convert2MethodRef
-        return callOnInstance(() -> Schema.instance.getVersion());
+        return Schema.instance.getVersion();
     }
 
-    public void schemaChange(String query)
+    public void startup()
     {
-        runOnInstance(() ->
+        throw new UnsupportedOperationException();
-        {
+    }
+
+    @Override
+    public void schemaChangeInternal(String query)
+    {
+        sync(() -> {
             try
             {
                 ClientState state = ClientState.forInternalCalls();
@@ -137,30 +164,33 @@ public class Instance extends InvokableInstance
             {
                 throw new RuntimeException("Error setting schema for test (query was: " + query + ")", e);
             }
-        });
+        }).run();
     }
 
-    private void registerMockMessaging(TestCluster cluster)
+    private void registerMockMessaging(ICluster cluster)
     {
-        BiConsumer<InetAddressAndPort, Message> deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message);
+        BiConsumer<InetAddressAndPort, IMessage> deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message);
-        BiConsumer<InetAddressAndPort, Message> deliverToInstanceIfNotFiltered = cluster.filters().filter(deliverToInstance);
+        BiConsumer<InetAddressAndPort, IMessage> deliverToInstanceIfNotFiltered = cluster.filters().filter(deliverToInstance);
 
         Map<InetAddress, InetAddressAndPort> addressAndPortMap = new HashMap<>();
-        cluster.stream().map(Instance::getBroadcastAddress).forEach(addressAndPort -> {
+        cluster.stream().forEach(instance -> {
-            if (null != addressAndPortMap.put(addressAndPort.address, addressAndPort))
+            InetAddressAndPort addressAndPort = instance.broadcastAddressAndPort();
-                throw new IllegalStateException("This version of Cassandra does not support multiple nodes with the same InetAddress");
+            if (!addressAndPort.equals(instance.config().broadcastAddressAndPort()))
+                throw new IllegalStateException("addressAndPort mismatch: " + addressAndPort + " vs " + instance.config().broadcastAddressAndPort());
+            InetAddressAndPort prev = addressAndPortMap.put(addressAndPort.address, addressAndPort);
+            if (null != prev)
+                throw new IllegalStateException("This version of Cassandra does not support multiple nodes with the same InetAddress: " + addressAndPort + " vs " + prev);
         });
 
-        acceptsOnInstance((BiConsumer<InetAddressAndPort, Message> deliver) ->
+        MessagingService.instance().addMessageSink(
-                MessagingService.instance().addMessageSink(new MessageDeliverySink(deliver, addressAndPortMap::get))
+                new MessageDeliverySink(deliverToInstanceIfNotFiltered, addressAndPortMap::get));
-        ).accept(deliverToInstanceIfNotFiltered);
     }
 
-    private static class MessageDeliverySink implements IMessageSink
+    private class MessageDeliverySink implements IMessageSink
     {
-        private final BiConsumer<InetAddressAndPort, Message> deliver;
+        private final BiConsumer<InetAddressAndPort, IMessage> deliver;
         private final Function<InetAddress, InetAddressAndPort> lookupAddressAndPort;
-        MessageDeliverySink(BiConsumer<InetAddressAndPort, Message> deliver, Function<InetAddress, InetAddressAndPort> lookupAddressAndPort)
+        MessageDeliverySink(BiConsumer<InetAddressAndPort, IMessage> deliver, Function<InetAddress, InetAddressAndPort> lookupAddressAndPort)
         {
             this.deliver = deliver;
             this.lookupAddressAndPort = lookupAddressAndPort;
@@ -170,7 +200,8 @@ public class Instance extends InvokableInstance
         {
             try (DataOutputBuffer out = new DataOutputBuffer(1024))
             {
-                InetAddressAndPort from = LegacyAdapter.getBroadcastAddressAndPort();
+                InetAddressAndPort from = broadcastAddressAndPort();
+                assert from.equals(lookupAddressAndPort.apply(messageOut.from));
                 InetAddressAndPort toFull = lookupAddressAndPort.apply(to);
                 messageOut.serialize(out, MessagingService.current_version);
                 deliver.accept(toFull, new Message(messageOut.verb.ordinal(), out.toByteArray(), id, MessagingService.current_version, from));
@@ -189,158 +220,146 @@ public class Instance extends InvokableInstance
         }
     }
 
-    private void receiveMessage(Message message)
+    public void receiveMessage(IMessage message)
     {
-        acceptsOnInstance((Message m) ->
+        sync(() -> {
-        {
+            try (DataInputBuffer in = new DataInputBuffer(message.bytes()))
-            try (DataInputBuffer in = new DataInputBuffer(m.bytes))
             {
-                MessageIn<?> messageIn = MessageIn.read(in, m.version, m.id);
+                MessageIn<?> messageIn = MessageIn.read(in, message.version(), message.id());
-                Runnable deliver = new MessageDeliveryTask(messageIn, m.id);
+                Runnable deliver = new MessageDeliveryTask(messageIn, message.id());
                 deliver.run();
             }
             catch (Throwable t)
             {
-                throw new RuntimeException("Exception occurred on the node " + LegacyAdapter.getBroadcastAddressAndPort(), t);
+                throw new RuntimeException("Exception occurred on node " + broadcastAddressAndPort(), t);
             }
-
+        }).run();
-        }).accept(message);
     }
 
-    void launch(TestCluster cluster)
+    @Override
+    public void startup(ICluster cluster)
     {
-        try
+        sync(() -> {
-        {
+            try
-            mkdirs();
+            {
-            int id = config.num;
+                mkdirs();
-            runOnInstance(() -> InstanceIDDefiner.instanceId = id); // for logging
+                Config.setOverrideLoadConfig(() -> loadConfig(config));
+                DatabaseDescriptor.daemonInitialization();
 
-            startup();
+                DatabaseDescriptor.createAllDirectories();
-            initializeRing(cluster);
+
-            registerMockMessaging(cluster);
+                // We need to persist this as soon as possible after startup checks.
-        }
+                // This should be the first write to SystemKeyspace (CASSANDRA-11742)
-        catch (Throwable t)
+                SystemKeyspace.persistLocalMetadata();
-        {
+                LegacySchemaMigrator.migrate();
-            if (t instanceof RuntimeException)
+
-                throw (RuntimeException) t;
+                try
-            throw new RuntimeException(t);
+                {
-        }
+                    // load schema from disk
+                    Schema.instance.loadFromDisk();
+                }
+                catch (Exception e)
+                {
+                    throw e;
+                }
+
+                Keyspace.setInitialized();
+
+                // Replay any CommitLogSegments found on disk
+                try
+                {
+                    CommitLog.instance.recoverSegmentsOnDisk();
+                }
+                catch (IOException e)
+                {
+                    throw new RuntimeException(e);
+                }
+
+                initializeRing(cluster);
+                registerMockMessaging(cluster);
+
+                SystemKeyspace.finishStartup();
+
+                if (!FBUtilities.getBroadcastAddress().equals(broadcastAddressAndPort().address))
+                    throw new IllegalStateException();
+                if (DatabaseDescriptor.getStoragePort() != broadcastAddressAndPort().port)
+                    throw new IllegalStateException();
+            }
+            catch (Throwable t)
+            {
+                if (t instanceof RuntimeException)
+                    throw (RuntimeException) t;
+                throw new RuntimeException(t);
+            }
+        }).run();
     }
 
     private void mkdirs()
     {
-        new File(config.saved_caches_directory).mkdirs();
+        new File(config.getString("saved_caches_directory")).mkdirs();
-        new File(config.hints_directory).mkdirs();
+        new File(config.getString("hints_directory")).mkdirs();
-        new File(config.commitlog_directory).mkdirs();
+        new File(config.getString("commitlog_directory")).mkdirs();
-        for (String dir : config.data_file_directories)
+        for (String dir : (String[]) config.get("data_file_directories"))
             new File(dir).mkdirs();
     }
 
-    private void startup()
+    private static Config loadConfig(IInstanceConfig overrides)
-    {
-        acceptsOnInstance((InstanceConfig config) ->
-        {
-            DatabaseDescriptor.daemonInitialization(() -> loadConfig(config));
-            DatabaseDescriptor.createAllDirectories();
-            Keyspace.setInitialized();
-            SystemKeyspace.persistLocalMetadata();
-        }).accept(config);
-    }
-
-
-    public static Config loadConfig(InstanceConfig overrides)
     {
         Config config = new Config();
-        // Defaults
+        overrides.propagate(config);
-        config.commitlog_sync = Config.CommitLogSync.batch;
-        config.endpoint_snitch = SimpleSnitch.class.getName();
-        config.seed_provider = new ParameterizedClass(SimpleSeedProvider.class.getName(),
-                                                      Collections.singletonMap("seeds", "127.0.0.1:7010"));
-        // Overrides
-        config.partitioner = overrides.partitioner;
-        config.broadcast_address = overrides.broadcast_address;
-        config.listen_address = overrides.listen_address;
-        config.broadcast_rpc_address = overrides.broadcast_rpc_address;
-        config.rpc_address = overrides.rpc_address;
-        config.saved_caches_directory = overrides.saved_caches_directory;
-        config.data_file_directories = overrides.data_file_directories;
-        config.commitlog_directory = overrides.commitlog_directory;
-        config.hints_directory = overrides.hints_directory;
-        config.cdc_raw_directory = overrides.cdc_directory;
-        config.concurrent_writes = overrides.concurrent_writes;
-        config.concurrent_counter_writes = overrides.concurrent_counter_writes;
-        config.concurrent_materialized_view_writes = overrides.concurrent_materialized_view_writes;
-        config.concurrent_reads = overrides.concurrent_reads;
-        config.memtable_flush_writers = overrides.memtable_flush_writers;
-        config.concurrent_compactors = overrides.concurrent_compactors;
-        config.memtable_heap_space_in_mb = overrides.memtable_heap_space_in_mb;
-        config.initial_token = overrides.initial_token;
-
-        // legacy config options we need to specify
-        config.commitlog_sync_batch_window_in_ms = 1.0;
         return config;
     }
 
-    private void initializeRing(TestCluster cluster)
+    private void initializeRing(ICluster cluster)
     {
         // This should be done outside instance in order to avoid serializing config
-        String partitionerName = config.partitioner;
+        String partitionerName = config.getString("partitioner");
         List<String> initialTokens = new ArrayList<>();
         List<InetAddressAndPort> hosts = new ArrayList<>();
         List<UUID> hostIds = new ArrayList<>();
         for (int i = 1 ; i <= cluster.size() ; ++i)
         {
-            InstanceConfig config = cluster.get(i).config;
+            IInstanceConfig config = cluster.get(i).config();
-            initialTokens.add(config.initial_token);
+            initialTokens.add(config.getString("initial_token"));
-            try
+            hosts.add(config.broadcastAddressAndPort());
-            {
+            hostIds.add(config.hostId());
-                hosts.add(InetAddressAndPort.getByName(config.broadcast_address));
-            }
-            catch (UnknownHostException e)
-            {
-                throw new RuntimeException(e);
-            }
-            hostIds.add(config.hostId);
         }
 
-        runOnInstance(() ->
+        try
         {
-            try
+            IPartitioner partitioner = FBUtilities.newPartitioner(partitionerName);
-            {
+            StorageService storageService = StorageService.instance;
-                IPartitioner partitioner = FBUtilities.newPartitioner(partitionerName);
+            List<Token> tokens = new ArrayList<>();
-                StorageService storageService = StorageService.instance;
+            for (String token : initialTokens)
-                List<Token> tokens = new ArrayList<>();
+                tokens.add(partitioner.getTokenFactory().fromString(token));
-                for (String token : initialTokens)
-                    tokens.add(partitioner.getTokenFactory().fromString(token));
 
-                for (int i = 0; i < tokens.size(); i++)
+            for (int i = 0; i < tokens.size(); i++)
-                {
-                    InetAddressAndPort ep = hosts.get(i);
-                    Gossiper.instance.initializeNodeUnsafe(ep.address, hostIds.get(i), 1);
-                    Gossiper.instance.injectApplicationState(ep.address,
-                            ApplicationState.TOKENS,
-                            new VersionedValue.VersionedValueFactory(partitioner).tokens(Collections.singleton(tokens.get(i))));
-                    storageService.onChange(ep.address,
-                            ApplicationState.STATUS,
-                            new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(tokens.get(i))));
-                    Gossiper.instance.realMarkAlive(ep.address, Gossiper.instance.getEndpointStateForEndpoint(ep.address));
-                    MessagingService.instance().setVersion(ep.address, MessagingService.current_version);
-                }
-
-                // check that all nodes are in token metadata
-                for (int i = 0; i < tokens.size(); ++i)
-                    assert storageService.getTokenMetadata().isMember(hosts.get(i).address);
-            }
-            catch (Throwable e) // UnknownHostException
             {
-                throw new RuntimeException(e);
+                InetAddressAndPort ep = hosts.get(i);
+                Gossiper.instance.initializeNodeUnsafe(ep.address, hostIds.get(i), 1);
+                Gossiper.instance.injectApplicationState(ep.address,
+                        ApplicationState.TOKENS,
+                        new VersionedValue.VersionedValueFactory(partitioner).tokens(Collections.singleton(tokens.get(i))));
+                storageService.onChange(ep.address,
+                        ApplicationState.STATUS,
+                        new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(tokens.get(i))));
+                Gossiper.instance.realMarkAlive(ep.address, Gossiper.instance.getEndpointStateForEndpoint(ep.address));
+                MessagingService.instance().setVersion(ep.address, MessagingService.current_version);
             }
-        });
+
+            // check that all nodes are in token metadata
+            for (int i = 0; i < tokens.size(); ++i)
+                assert storageService.getTokenMetadata().isMember(hosts.get(i).address);
+        }
+        catch (Throwable e) // UnknownHostException
+        {
+            throw new RuntimeException(e);
+        }
     }
 
-    void shutdown()
+    public void shutdown()
     {
-        acceptsOnInstance((ExecutorService executor) -> {
+        sync((ExecutorService executor) -> {
             Throwable error = null;
             error = parallelRun(error, executor,
                     Gossiper.instance::stop,
@@ -369,9 +388,9 @@ public class Instance extends InvokableInstance
             );
             LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
             loggerContext.stop();
+            super.shutdown();
             Throwables.maybeFail(error);
         }).accept(isolatedExecutor);
-        super.shutdown();
     }
 
     private static Throwable parallelRun(Throwable accumulate, ExecutorService runOn, ThrowingRunnable ... runnables)

test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java

@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import com.google.common.base.Predicate;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.utils.Pair;
+
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.util.Arrays;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class InstanceClassLoader extends URLClassLoader
+{
+    // Classes that have to be shared between instances, for configuration or returning values
+    private static final Set<String> sharedClassNames = Arrays.stream(new Class[]
+            {
+                    Pair.class,
+                    InetAddressAndPort.class,
+                    ParameterizedClass.class,
+                    IInvokableInstance.class
+            })
+            .map(Class::getName)
+            .collect(Collectors.toSet());
+
+    private static final Predicate<String> sharePackage = name ->
+               name.startsWith("org.apache.cassandra.distributed.api.")
+            || name.startsWith("sun.")
+            || name.startsWith("oracle.")
+            || name.startsWith("com.sun.")
+            || name.startsWith("com.oracle.")
+            || name.startsWith("java.")
+            || name.startsWith("javax.")
+            || name.startsWith("jdk.")
+            || name.startsWith("netscape.")
+            || name.startsWith("org.xml.sax.");
+
+    private static final Predicate<String> shareClass = name -> sharePackage.apply(name) || sharedClassNames.contains(name);
+
+    public static interface Factory
+    {
+        InstanceClassLoader create(int id, URL[] urls, ClassLoader sharedClassLoader);
+    }
+
+    private final ClassLoader sharedClassLoader;
+
+    InstanceClassLoader(int id, URL[] urls, ClassLoader sharedClassLoader)
+    {
+        super(urls, null);
+        this.sharedClassLoader = sharedClassLoader;
+    }
+
+    @Override
+    public Class<?> loadClass(String name) throws ClassNotFoundException
+    {
+        if (shareClass.apply(name))
+            return sharedClassLoader.loadClass(name);
+
+        return loadClassInternal(name);
+    }
+
+    Class<?> loadClassInternal(String name) throws ClassNotFoundException
+    {
+        synchronized (getClassLoadingLock(name))
+        {
+            // First, check if the class has already been loaded
+            Class<?> c = findLoadedClass(name);
+
+            if (c == null)
+                c = findClass(name);
+
+            return c;
+        }
+    }
+
+    /**
+     * @return true iff this class was loaded by an InstanceClassLoader, and as such is used by a dtest node
+     */
+    public static boolean wasLoadedByAnInstanceClassLoader(Class<?> clazz)
+    {
+        return clazz.getClassLoader().getClass().getName().equals(InstanceClassLoader.class.getName());
+    }
+
+}

test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java

@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.SimpleSeedProvider;
+import org.apache.cassandra.locator.SimpleSnitch;
+
+import java.io.File;
+import java.lang.reflect.Field;
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.UUID;
+
+public class InstanceConfig implements IInstanceConfig
+{
+    private static final Object NULL = new Object();
+
+    public final int num;
+    public int num() { return num; }
+
+    public final UUID hostId;
+    public UUID hostId() { return hostId; }
+    private final Map<String, Object> params = new TreeMap<>();
+
+    private volatile InetAddressAndPort broadcastAddressAndPort;
+
+    @Override
+    public InetAddressAndPort broadcastAddressAndPort()
+    {
+        if (broadcastAddressAndPort == null)
+        {
+            try
+            {
+                broadcastAddressAndPort = InetAddressAndPort.getByNameOverrideDefaults(getString("broadcast_address"), getInt("storage_port"));
+            }
+            catch (UnknownHostException e)
+            {
+                throw new IllegalStateException(e);
+            }
+        }
+        return broadcastAddressAndPort;
+    }
+
+    private InstanceConfig(int num,
+                           String broadcast_address,
+                           String listen_address,
+                           String broadcast_rpc_address,
+                           String rpc_address,
+                           String saved_caches_directory,
+                           String[] data_file_directories,
+                           String commitlog_directory,
+                           String hints_directory,
+                           String cdc_raw_directory,
+                           String initial_token)
+    {
+        this.num = num;
+        this.hostId = java.util.UUID.randomUUID();
+        this    .set("broadcast_address", broadcast_address)
+                .set("listen_address", listen_address)
+                .set("broadcast_rpc_address", broadcast_rpc_address)
+                .set("rpc_address", rpc_address)
+                .set("saved_caches_directory", saved_caches_directory)
+                .set("data_file_directories", data_file_directories)
+                .set("commitlog_directory", commitlog_directory)
+                .set("hints_directory", hints_directory)
+                .set("cdc_raw_directory", cdc_raw_directory)
+                .set("initial_token", initial_token)
+                .set("partitioner", "org.apache.cassandra.dht.Murmur3Partitioner")
+                .set("concurrent_writes", 2)
+                .set("concurrent_counter_writes", 2)
+                .set("concurrent_materialized_view_writes", 2)
+                .set("concurrent_reads", 2)
+                .set("memtable_flush_writers", 1)
+                .set("concurrent_compactors", 1)
+                .set("memtable_heap_space_in_mb", 10)
+                .set("commitlog_sync", "batch")
+                .set("storage_port", 7010)
+                .set("endpoint_snitch", SimpleSnitch.class.getName())
+                .set("seed_provider", new ParameterizedClass(SimpleSeedProvider.class.getName(),
+                        Collections.singletonMap("seeds", "127.0.0.1")))
+                // legacy parameters
+                .forceSet("commitlog_sync_batch_window_in_ms", 1.0);
+    }
+
+    private InstanceConfig(InstanceConfig copy)
+    {
+        this.num = copy.num;
+        this.params.putAll(copy.params);
+        this.hostId = copy.hostId;
+    }
+
+    public InstanceConfig set(String fieldName, Object value)
+    {
+        if (value == null)
+            value = NULL;
+
+        // test value
+        propagate(new Config(), fieldName, value, false);
+        params.put(fieldName, value);
+        return this;
+    }
+
+    private InstanceConfig forceSet(String fieldName, Object value)
+    {
+        if (value == null)
+            value = NULL;
+
+        // test value
+        params.put(fieldName, value);
+        return this;
+    }
+
+    public void propagateIfSet(Object writeToConfig, String fieldName)
+    {
+        if (params.containsKey(fieldName))
+            propagate(writeToConfig, fieldName, params.get(fieldName), true);
+    }
+
+    public void propagate(Object writeToConfig)
+    {
+        for (Map.Entry<String, Object> e : params.entrySet())
+            propagate(writeToConfig, e.getKey(), e.getValue(), true);
+    }
+
+    private void propagate(Object writeToConfig, String fieldName, Object value, boolean ignoreMissing)
+    {
+        if (value == NULL)
+            value = null;
+
+        Class<?> configClass = writeToConfig.getClass();
+        Field valueField;
+        try
+        {
+            valueField = configClass.getDeclaredField(fieldName);
+        }
+        catch (NoSuchFieldException e)
+        {
+            if (!ignoreMissing)
+                throw new IllegalStateException(e);
+            return;
+        }
+
+        if (valueField.getType().isEnum() && value instanceof String)
+        {
+            String test = (String) value;
+            value = Arrays.stream(valueField.getType().getEnumConstants())
+                    .filter(e -> ((Enum<?>)e).name().equals(test))
+                    .findFirst()
+                    .get();
+        }
+        try
+        {
+            valueField.set(writeToConfig, value);
+        }
+        catch (IllegalAccessException e)
+        {
+            throw new IllegalStateException(e);
+        }
+        catch (IllegalArgumentException e)
+        {
+            throw new IllegalStateException(e);
+        }
+    }
+
+    public Object get(String name)
+    {
+        return params.get(name);
+    }
+
+    public int getInt(String name)
+    {
+        return (Integer)params.get(name);
+    }
+
+    public String getString(String name)
+    {
+        return (String)params.get(name);
+    }
+
+    public static InstanceConfig generate(int nodeNum, File root, String token)
+    {
+        return new InstanceConfig(nodeNum,
+                                  "127.0.0." + nodeNum,
+                                  "127.0.0." + nodeNum,
+                                  "127.0.0." + nodeNum,
+                                  "127.0.0." + nodeNum,
+                                  String.format("%s/node%d/saved_caches", root, nodeNum),
+                                  new String[] { String.format("%s/node%d/data", root, nodeNum) },
+                                  String.format("%s/node%d/commitlog", root, nodeNum),
+                                  String.format("%s/node%d/hints", root, nodeNum),
+                                  String.format("%s/node%d/cdc", root, nodeNum),
+                                  token);
+    }
+
+    public InstanceConfig forVersion(Versions.Major major)
+    {
+        switch (major)
+        {
+            case v4: return this;
+            default: return new InstanceConfig(this)
+                            .set("seed_provider", new ParameterizedClass(SimpleSeedProvider.class.getName(),
+                                                                         Collections.singletonMap("seeds", "127.0.0.1")));
+        }
+    }
+
+    public String toString()
+    {
+        return params.toString();
+    }
+}

test/distributed/org/apache/cassandra/distributed/impl/InstanceIDDefiner.java

@@ -16,9 +16,10 @@
  * limitations under the License.
  */
 
-package org.apache.cassandra.distributed;
+package org.apache.cassandra.distributed.impl;
 
 import ch.qos.logback.core.PropertyDefinerBase;
+import org.apache.cassandra.concurrent.NamedThreadFactory;
 
 /**
  * Used by logback to find/define property value, see logback-dtest.xml
@@ -26,13 +27,15 @@ import ch.qos.logback.core.PropertyDefinerBase;
 public class InstanceIDDefiner extends PropertyDefinerBase
 {
     // Instantiated per classloader, set by Instance
-    public static int instanceId = -1;
+    private static volatile String instanceId = "<main>";
+    public static void setInstanceId(int id)
+    {
+        instanceId = "node" + id;
+        NamedThreadFactory.setGlobalPrefix("node" + id + "_");
+    }
 
     public String getPropertyValue()
     {
-        if (instanceId == -1)
+        return instanceId;
-            return "<main>";
-        else
-            return "INSTANCE" + instanceId;
     }
 }

test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java

@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.apache.cassandra.concurrent.NamedThreadFactory;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.utils.Throwables;
+
+public class IsolatedExecutor implements IIsolatedExecutor
+{
+    final ExecutorService isolatedExecutor;
+    private final ClassLoader classLoader;
+    private final Method deserializeOnInstance;
+
+    IsolatedExecutor(String name, ClassLoader classLoader)
+    {
+        this.isolatedExecutor = Executors.newCachedThreadPool(new NamedThreadFactory("isolatedExecutor", Thread.NORM_PRIORITY, classLoader, new ThreadGroup(name)));
+        this.classLoader = classLoader;
+        this.deserializeOnInstance = lookupDeserializeOneObject(classLoader);
+    }
+
+    public void shutdown()
+    {
+        isolatedExecutor.shutdown();
+    }
+
+    public <O> CallableNoExcept<Future<O>> async(CallableNoExcept<O> call) { return () -> isolatedExecutor.submit(call); }
+    public <O> CallableNoExcept<O> sync(CallableNoExcept<O> call) { return () -> waitOn(async(call).call()); }
+
+    public CallableNoExcept<Future<?>> async(Runnable run) { return () -> isolatedExecutor.submit(run); }
+    public Runnable sync(Runnable run) { return () -> waitOn(async(run).call()); }
+
+    public <I> Function<I, Future<?>> async(Consumer<I> consumer) { return (a) -> isolatedExecutor.submit(() -> consumer.accept(a)); }
+    public <I> Consumer<I> sync(Consumer<I> consumer) { return (a) -> waitOn(async(consumer).apply(a)); }
+
+    public <I1, I2> BiFunction<I1, I2, Future<?>> async(BiConsumer<I1, I2> consumer) { return (a, b) -> isolatedExecutor.submit(() -> consumer.accept(a, b)); }
+    public <I1, I2> BiConsumer<I1, I2> sync(BiConsumer<I1, I2> consumer) { return (a, b) -> waitOn(async(consumer).apply(a, b)); }
+
+    public <I, O> Function<I, Future<O>> async(Function<I, O> f) { return (a) -> isolatedExecutor.submit(() -> f.apply(a)); }
+    public <I, O> Function<I, O> sync(Function<I, O> f) { return (a) -> waitOn(async(f).apply(a)); }
+
+    public <I1, I2, O> BiFunction<I1, I2, Future<O>> async(BiFunction<I1, I2, O> f) { return (a, b) -> isolatedExecutor.submit(() -> f.apply(a, b)); }
+    public <I1, I2, O> BiFunction<I1, I2, O> sync(BiFunction<I1, I2, O> f) { return (a, b) -> waitOn(async(f).apply(a, b)); }
+
+    public <I1, I2, I3, O> TriFunction<I1, I2, I3, Future<O>> async(TriFunction<I1, I2, I3, O> f) { return (a, b, c) -> isolatedExecutor.submit(() -> f.apply(a, b, c)); }
+    public <I1, I2, I3, O> TriFunction<I1, I2, I3, O> sync(TriFunction<I1, I2, I3, O> f) { return (a, b, c) -> waitOn(async(f).apply(a, b, c)); }
+
+    public <E extends Serializable> E transfer(E object)
+    {
+        return (E) transferOneObject(object, classLoader, deserializeOnInstance);
+    }
+
+    static <E extends Serializable> E transferAdhoc(E object, ClassLoader classLoader)
+    {
+        return transferOneObject(object, classLoader, lookupDeserializeOneObject(classLoader));
+    }
+
+    private static <E extends Serializable> E transferOneObject(E object, ClassLoader classLoader, Method deserializeOnInstance)
+    {
+        byte[] bytes = serializeOneObject(object);
+        try
+        {
+            Object onInstance = deserializeOnInstance.invoke(null, bytes);
+            if (onInstance.getClass().getClassLoader() != classLoader)
+                throw new IllegalStateException(onInstance + " seemingly from wrong class loader: " + onInstance.getClass().getClassLoader() + ", but expected " + classLoader);
+
+            return (E) onInstance;
+        }
+        catch (IllegalAccessException | InvocationTargetException e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static Method lookupDeserializeOneObject(ClassLoader classLoader)
+    {
+        try
+        {
+            return classLoader.loadClass(IsolatedExecutor.class.getName()).getDeclaredMethod("deserializeOneObject", byte[].class);
+        }
+        catch (ClassNotFoundException | NoSuchMethodException e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @SuppressWarnings("unused") // called through method invocation
+    public static Object deserializeOneObject(byte[] bytes)
+    {
+        try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+             ObjectInputStream ois = new ObjectInputStream(bais);)
+        {
+            return ois.readObject();
+        }
+        catch (IOException | ClassNotFoundException e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static byte[] serializeOneObject(Object object)
+    {
+        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+             ObjectOutputStream oos = new ObjectOutputStream(baos))
+        {
+            oos.writeObject(object);
+            oos.close();
+            return baos.toByteArray();
+        }
+        catch (IOException e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static <T> T waitOn(Future<T> f)
+    {
+        try
+        {
+            return f.get();
+        }
+        catch (InterruptedException e)
+        {
+            throw new RuntimeException(e);
+        }
+        catch (ExecutionException e)
+        {
+            throw new RuntimeException(e.getCause());
+        }
+    }
+
+}

test/distributed/org/apache/cassandra/distributed/impl/Listen.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.LockSupport;
+import org.apache.cassandra.distributed.api.IListen;
+
+public class Listen implements IListen
+{
+    final Instance instance;
+    public Listen(Instance instance)
+    {
+        this.instance = instance;
+    }
+
+    public Cancel schema(Runnable onChange)
+    {
+        final AtomicBoolean cancel = new AtomicBoolean();
+        instance.isolatedExecutor.execute(() -> {
+            UUID prev = instance.schemaVersion();
+            while (true)
+            {
+                if (cancel.get())
+                    return;
+
+                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10L));
+
+                UUID cur = instance.schemaVersion();
+                if (!prev.equals(cur))
+                    onChange.run();
+                prev = cur;
+            }
+        });
+        return () -> cancel.set(true);
+    }
+}

test/distributed/org/apache/cassandra/distributed/impl/Message.java

@@ -16,18 +16,19 @@
  * limitations under the License.
  */
 
-package org.apache.cassandra.distributed;
+package org.apache.cassandra.distributed.impl;
 
+import org.apache.cassandra.distributed.api.IMessage;
 import org.apache.cassandra.locator.InetAddressAndPort;
 
 // a container for simplifying the method signature for per-instance message handling/delivery
-public class Message
+public class Message implements IMessage
 {
-    public final int verb;
+    private final int verb;
-    public final byte[] bytes;
+    private final byte[] bytes;
-    public final int id;
+    private final int id;
-    public final int version;
+    private final int version;
-    public final InetAddressAndPort from;
+    private final InetAddressAndPort from;
 
     public Message(int verb, byte[] bytes, int id, int version, InetAddressAndPort from)
     {
@@ -37,5 +38,35 @@ public class Message
         this.version = version;
         this.from = from;
     }
+
+    @Override
+    public int verb()
+    {
+        return verb;
+    }
+
+    @Override
+    public byte[] bytes()
+    {
+        return bytes;
+    }
+
+    @Override
+    public int id()
+    {
+        return id;
+    }
+
+    @Override
+    public int version()
+    {
+        return version;
+    }
+
+    @Override
+    public InetAddressAndPort from()
+    {
+        return from;
+    }
 }
 

test/distributed/org/apache/cassandra/distributed/impl/MessageFilters.java

@@ -16,36 +16,44 @@
  * limitations under the License.
  */
 
-package org.apache.cassandra.distributed;
+package org.apache.cassandra.distributed.impl;
 
 import java.util.Arrays;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.function.BiConsumer;
 
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IMessage;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.ICluster;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.net.MessagingService;
 
-public class MessageFilters
+public class MessageFilters implements IMessageFilters
 {
-    private final TestCluster cluster;
+    private final ICluster cluster;
     private final Set<Filter> filters = new CopyOnWriteArraySet<>();
 
-    public MessageFilters(TestCluster cluster)
+    public MessageFilters(AbstractCluster cluster)
     {
         this.cluster = cluster;
     }
 
-    BiConsumer<InetAddressAndPort, Message> filter(BiConsumer<InetAddressAndPort, Message> applyIfNotFiltered)
+    public BiConsumer<InetAddressAndPort, IMessage> filter(BiConsumer<InetAddressAndPort, IMessage> applyIfNotFiltered)
     {
         return (toAddress, message) ->
         {
-            int from = cluster.get(message.from).config.num;
+            IInstance from = cluster.get(message.from());
-            int to = cluster.get(toAddress).config.num;
+            IInstance to = cluster.get(toAddress);
-            int verb = message.verb;
+            if (from == null || to == null)
+                return; // cannot deliver
+            int fromNum = from.config().num();
+            int toNum = to.config().num();
+            int verb = message.verb();
             for (Filter filter : filters)
             {
-                if (filter.matches(from, to, verb))
+                if (filter.matches(fromNum, toNum, verb))
                     return;
             }
 
@@ -53,7 +61,7 @@ public class MessageFilters
         };
     }
 
-    public class Filter
+    public class Filter implements IMessageFilters.Filter
     {
         final int[] from;
         final int[] to;
@@ -120,7 +128,7 @@ public class MessageFilters
         }
     }
 
-    public class Builder
+    public class Builder implements IMessageFilters.Builder
     {
         int[] from;
         int[] to;
@@ -154,7 +162,8 @@ public class MessageFilters
         }
     }
 
-    public Builder verbs(MessagingService.Verb ... verbs)
+    @Override
+    public Builder verbs(MessagingService.Verb... verbs)
     {
         int[] ids = new int[verbs.length];
         for (int i = 0 ; i < verbs.length ; ++i)
@@ -162,11 +171,13 @@ public class MessageFilters
         return new Builder(ids);
     }
 
+    @Override
     public Builder allVerbs()
     {
         return new Builder(null);
     }
 
+    @Override
     public void reset()
     {
         filters.clear();

test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.cassandra.distributed;
+package org.apache.cassandra.distributed.impl;
 
 import java.nio.ByteBuffer;
 import java.util.List;

test/distributed/org/apache/cassandra/distributed/impl/Versions.java

@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import java.io.File;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.utils.FBUtilities;
+
+public class Versions
+{
+    private static final Logger logger = LoggerFactory.getLogger(Versions.class);
+    public static Version CURRENT = new Version(FBUtilities.getReleaseVersionString(), ((URLClassLoader)Versions.class.getClassLoader()).getURLs());
+
+    public enum Major
+    {
+        v22("2\\.2\\.([0-9]+)"),
+        v30("3\\.0\\.([0-9]+)"),
+        v3X("3\\.([1-9]|1[01])(\\.([0-9]+))?"),
+        v4("4\\.([0-9]+)");
+        final Pattern pattern;
+        Major(String verify)
+        {
+            this.pattern = Pattern.compile(verify);
+        }
+
+        static Major fromFull(String version)
+        {
+            if (version.isEmpty())
+                throw new IllegalArgumentException(version);
+            switch (version.charAt(0))
+            {
+                case '2':
+                    if (version.startsWith("2.2"))
+                        return v22;
+                    throw new IllegalArgumentException(version);
+                case '3':
+                    if (version.startsWith("3.0"))
+                        return v30;
+                    return v3X;
+                case '4':
+                    return v4;
+                default:
+                    throw new IllegalArgumentException(version);
+            }
+        }
+
+        // verify that the version string is valid for this major version
+        boolean verify(String version)
+        {
+            return pattern.matcher(version).matches();
+        }
+
+        // sort two strings of the same major version as this enum instance
+        int compare(String a, String b)
+        {
+            Matcher ma = pattern.matcher(a);
+            Matcher mb = pattern.matcher(a);
+            if (!ma.matches()) throw new IllegalArgumentException(a);
+            if (!mb.matches()) throw new IllegalArgumentException(b);
+            int result = Integer.compare(Integer.parseInt(ma.group(1)), Integer.parseInt(mb.group(1)));
+            if (result == 0 && this == v3X && (ma.group(3) != null || mb.group(3) != null))
+            {
+                if (ma.group(3) != null && mb.group(3) != null)
+                {
+                    result = Integer.compare(Integer.parseInt(ma.group(3)), Integer.parseInt(mb.group(3)));
+                }
+                else
+                {
+                    result = ma.group(3) != null ? 1 : -1;
+                }
+            }
+            // sort descending
+            return -result;
+        }
+    }
+
+    public static class Version
+    {
+        public final Major major;
+        public final String version;
+        public final URL[] classpath;
+
+        public Version(String version, URL[] classpath)
+        {
+            this(Major.fromFull(version), version, classpath);
+        }
+        public Version(Major major, String version, URL[] classpath)
+        {
+            this.major = major;
+            this.version = version;
+            this.classpath = classpath;
+        }
+    }
+
+    private final Map<Major, List<Version>> versions;
+    public Versions(Map<Major, List<Version>> versions)
+    {
+        this.versions = versions;
+    }
+
+    public Version get(String full)
+    {
+        return versions.get(Major.fromFull(full))
+                       .stream()
+                       .filter(v -> full.equals(v.version))
+                       .findFirst()
+                       .orElseThrow(() -> new RuntimeException("No version " + full + " found"));
+    }
+
+    public Version getLatest(Major major)
+    {
+        return versions.get(major).stream().findFirst().orElseThrow(() -> new RuntimeException("No " + major + " versions found"));
+    }
+
+    public static Versions find()
+    {
+        logger.info("Looking for dtest jars in " + new File("build").getAbsolutePath());
+        final Pattern pattern = Pattern.compile("dtest-([0-9.]+)\\.jar");
+        final Map<Major, List<Version>> versions = new HashMap<>();
+        for (Major major : Major.values())
+            versions.put(major, new ArrayList<>());
+
+        for (File file : new File("build").listFiles())
+        {
+            Matcher m = pattern.matcher(file.getName());
+            if (!m.matches())
+                continue;
+            String version = m.group(1);
+            Major major = Major.fromFull(version);
+            versions.get(major).add(new Version(major, version, new URL[] { toURL(file) }));
+        }
+
+        for (Map.Entry<Major, List<Version>> e : versions.entrySet())
+        {
+            if (e.getValue().isEmpty())
+                continue;
+            Collections.sort(e.getValue(), Comparator.comparing(v -> v.version, e.getKey()::compare));
+            logger.info("Found " + e.getValue().stream().map(v -> v.version).collect(Collectors.joining(", ")));
+        }
+
+        return new Versions(versions);
+    }
+
+    public static URL toURL(File file)
+    {
+        try
+        {
+            return file.toURI().toURL();
+        }
+        catch (MalformedURLException e)
+        {
+            throw new IllegalArgumentException(e);
+        }
+    }
+
+}

test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java

@@ -16,19 +16,20 @@
  * limitations under the License.
  */
 
-package org.apache.cassandra.distributed;
+package org.apache.cassandra.distributed.test;
 
 import org.junit.Assert;
 import org.junit.Test;
 
 import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.distributed.Cluster;
 
 public class DistributedReadWritePathTest extends DistributedTestBase
 {
     @Test
     public void coordinatorRead() throws Throwable
     {
-        try (TestCluster cluster = createCluster(3))
+        try (Cluster cluster = init(Cluster.create(3)))
         {
             cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
 
@@ -36,7 +37,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase
             cluster.get(2).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 2, 2)");
             cluster.get(3).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 3, 3)");
 
-            assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?",
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?",
                                                      ConsistencyLevel.ALL,
                                                      1),
                        row(1, 1, 1),
@@ -48,11 +49,11 @@ public class DistributedReadWritePathTest extends DistributedTestBase
     @Test
     public void coordinatorWrite() throws Throwable
     {
-        try (TestCluster cluster = createCluster(3))
+        try (Cluster cluster = init(Cluster.create(3)))
         {
             cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
 
-            cluster.coordinator().execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)",
+            cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)",
                                           ConsistencyLevel.QUORUM);
 
             for (int i = 0; i < 3; i++)
@@ -61,7 +62,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase
                            row(1, 1, 1));
             }
 
-            assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
                                                      ConsistencyLevel.QUORUM),
                        row(1, 1, 1));
         }
@@ -70,7 +71,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase
     @Test
     public void readRepairTest() throws Throwable
     {
-        try (TestCluster cluster = createCluster(3))
+        try (Cluster cluster = init(Cluster.create(3)))
         {
             cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
 
@@ -79,7 +80,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase
 
             assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"));
 
-            assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
                                                      ConsistencyLevel.QUORUM),
                        row(1, 1, 1));
 
@@ -92,7 +93,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase
     @Test
     public void writeWithSchemaDisagreement() throws Throwable
     {
-        try (TestCluster cluster = createCluster(3))
+        try (Cluster cluster = init(Cluster.create(3)))
         {
             cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v1 int, PRIMARY KEY (pk, ck))");
 
@@ -106,7 +107,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase
             Exception thrown = null;
             try
             {
-                cluster.coordinator().execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v1, v2) VALUES (2, 2, 2, 2)",
+                cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v1, v2) VALUES (2, 2, 2, 2)",
                                               ConsistencyLevel.QUORUM);
             }
             catch (RuntimeException e)
@@ -114,15 +115,15 @@ public class DistributedReadWritePathTest extends DistributedTestBase
                 thrown = e;
             }
 
-            Assert.assertTrue(thrown.getMessage().contains("Exception occurred on the node"));
+            Assert.assertTrue(thrown.getMessage().contains("Exception occurred on node"));
-            Assert.assertTrue(thrown.getCause().getMessage().contains("Unknown column v2 during deserialization"));
+            Assert.assertTrue(thrown.getCause().getCause().getCause().getMessage().contains("Unknown column v2 during deserialization"));
         }
     }
 
     @Test
     public void readWithSchemaDisagreement() throws Throwable
     {
-        try (TestCluster cluster = createCluster(3))
+        try (Cluster cluster = init(Cluster.create(3)))
         {
             cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v1 int, PRIMARY KEY (pk, ck))");
 
@@ -136,7 +137,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase
             Exception thrown = null;
             try
             {
-                assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
+                assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
                                                          ConsistencyLevel.ALL),
                            row(1, 1, 1, null));
             }
@@ -144,8 +145,8 @@ public class DistributedReadWritePathTest extends DistributedTestBase
             {
                 thrown = e;
             }
-            Assert.assertTrue(thrown.getMessage().contains("Exception occurred on the node"));
+            Assert.assertTrue(thrown.getMessage().contains("Exception occurred on node"));
-            Assert.assertTrue(thrown.getCause().getMessage().contains("Unknown column v2 during deserialization"));
+            Assert.assertTrue(thrown.getCause().getCause().getCause().getMessage().contains("Unknown column v2 during deserialization"));
         }
     }
 }

test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java

@@ -16,16 +16,18 @@
  * limitations under the License.
  */
 
-package org.apache.cassandra.distributed;
+package org.apache.cassandra.distributed.test;
 
 import java.util.Arrays;
 
 import org.junit.Assert;
 import org.junit.BeforeClass;
 
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+
 public class DistributedTestBase
 {
-    static String KEYSPACE = "distributed_test_keyspace";
+    public static String KEYSPACE = "distributed_test_keyspace";
 
     @BeforeClass
     public static void setup()
@@ -33,11 +35,9 @@ public class DistributedTestBase
         System.setProperty("org.apache.cassandra.disable_mbean_registration", "true");
     }
 
-    TestCluster createCluster(int nodeCount) throws Throwable
+    protected static <C extends AbstractCluster<?>> C init(C cluster)
     {
-        TestCluster cluster = TestCluster.create(nodeCount);
+        cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + cluster.size() + "};");
-        cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + nodeCount + "};");
-
         return cluster;
     }
 

test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTest.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.upgrade;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.distributed.impl.Versions;
+import org.apache.cassandra.distributed.test.DistributedTestBase;
+
+public class UpgradeTest extends UpgradeTestBase
+{
+
+    @Test
+    public void upgradeTest() throws Throwable
+    {
+        new TestCase()
+            .upgrade(Versions.Major.v22, Versions.Major.v30, Versions.Major.v3X)
+            .setup((cluster) -> {
+                cluster.schemaChange("CREATE TABLE " + DistributedTestBase.KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+                cluster.get(1).executeInternal("INSERT INTO " + DistributedTestBase.KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)");
+                cluster.get(2).executeInternal("INSERT INTO " + DistributedTestBase.KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 2, 2)");
+                cluster.get(3).executeInternal("INSERT INTO " + DistributedTestBase.KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 3, 3)");
+            })
+            .runAfterClusterUpgrade((cluster) -> {
+                DistributedTestBase.assertRows(cluster.coordinator(1).execute("SELECT * FROM " + DistributedTestBase.KEYSPACE + ".tbl WHERE pk = ?",
+                                                                              ConsistencyLevel.ALL,
+                                                                              1),
+                                               DistributedTestBase.row(1, 1, 1),
+                                               DistributedTestBase.row(1, 2, 2),
+                                               DistributedTestBase.row(1, 3, 3));
+            }).run();
+    }
+
+}

test/distributed/org/apache/cassandra/distributed/upgrade/UpgradeTestBase.java

@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.upgrade;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.impl.Instance;
+import org.apache.cassandra.distributed.impl.Versions;
+import org.apache.cassandra.distributed.impl.Versions.Version;
+import org.apache.cassandra.distributed.test.DistributedTestBase;
+
+import static org.apache.cassandra.distributed.impl.Versions.*;
+
+public class UpgradeTestBase extends DistributedTestBase
+{
+    public static interface RunOnCluster
+    {
+        public void run(UpgradeableCluster cluster) throws Throwable;
+    }
+
+    public static interface RunOnClusterAndNode
+    {
+        public void run(UpgradeableCluster cluster, int node) throws Throwable;
+    }
+
+    public static class TestVersions
+    {
+        final Version initial;
+        final Version[] upgrade;
+
+        public TestVersions(Version initial, Version ... upgrade)
+        {
+            this.initial = initial;
+            this.upgrade = upgrade;
+        }
+    }
+
+    public static class TestCase implements Instance.ThrowingRunnable
+    {
+        private final Versions versions;
+        private final List<TestVersions> upgrade = new ArrayList<>();
+        private int nodeCount = 3;
+        private RunOnCluster setup;
+        private RunOnClusterAndNode runAfterNodeUpgrade;
+        private RunOnCluster runAfterClusterUpgrade;
+
+        public TestCase()
+        {
+            this(find());
+        }
+
+        public TestCase(Versions versions)
+        {
+            this.versions = versions;
+        }
+
+        public TestCase nodes(int nodeCount)
+        {
+            this.nodeCount = nodeCount;
+            return this;
+        }
+
+        public TestCase upgrade(Major initial, Major ... upgrade)
+        {
+            this.upgrade.add(new TestVersions(versions.getLatest(initial),
+                                              Arrays.stream(upgrade)
+                                                    .map(versions::getLatest)
+                                                    .toArray(Version[]::new)));
+            return this;
+        }
+
+        public TestCase upgrade(Version initial, Version ... upgrade)
+        {
+            this.upgrade.add(new TestVersions(initial, upgrade));
+            return this;
+        }
+
+        public TestCase setup(RunOnCluster setup)
+        {
+            this.setup = setup;
+            return this;
+        }
+
+        public TestCase runAfterNodeUpgrade(RunOnClusterAndNode runAfterNodeUpgrade)
+        {
+            this.runAfterNodeUpgrade = runAfterNodeUpgrade;
+            return this;
+        }
+
+        public TestCase runAfterClusterUpgrade(RunOnCluster runAfterClusterUpgrade)
+        {
+            this.runAfterClusterUpgrade = runAfterClusterUpgrade;
+            return this;
+        }
+
+        public void run() throws Throwable
+        {
+            if (setup == null)
+                throw new AssertionError();
+            if (upgrade.isEmpty())
+                throw new AssertionError();
+            if (runAfterClusterUpgrade == null && runAfterNodeUpgrade == null)
+                throw new AssertionError();
+            if (runAfterClusterUpgrade == null)
+                runAfterClusterUpgrade = (c) -> {};
+            if (runAfterNodeUpgrade == null)
+                runAfterNodeUpgrade = (c, n) -> {};
+
+            for (TestVersions upgrade : this.upgrade)
+            {
+                try (UpgradeableCluster cluster = init(UpgradeableCluster.create(nodeCount, upgrade.initial)))
+                {
+                    cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+                    cluster.get(1).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)");
+                    cluster.get(2).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 2, 2)");
+                    cluster.get(3).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 3, 3)");
+
+                    for (Version version : upgrade.upgrade)
+                    {
+                        for (int n = 1 ; n <= nodeCount ; ++n)
+                        {
+                            cluster.get(n).shutdown();
+                            cluster.get(n).setVersion(version);
+                            cluster.get(n).startup();
+                            runAfterNodeUpgrade.run(cluster, n);
+                        }
+
+                        runAfterClusterUpgrade.run(cluster);
+                    }
+                }
+
+            }
+        }
+    }
+
+}

test/unit/org/apache/cassandra/LogbackStatusListener.java

@@ -33,7 +33,7 @@ import ch.qos.logback.classic.LoggerContext;
 import ch.qos.logback.classic.spi.LoggerContextListener;
 import ch.qos.logback.core.status.Status;
 import ch.qos.logback.core.status.StatusListener;
-import org.apache.cassandra.distributed.InstanceClassLoader;
+import org.apache.cassandra.distributed.impl.InstanceClassLoader;
 
 /*
  * Listen for logback readiness and then redirect stdout/stderr to logback

test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java

@@ -83,20 +83,20 @@ public class DatabaseDescriptorRefTest
     "org.apache.cassandra.config.YamlConfigurationLoader$CustomConstructor",
     "org.apache.cassandra.config.TransparentDataEncryptionOptions",
     "org.apache.cassandra.dht.IPartitioner",
-    "org.apache.cassandra.distributed.InstanceClassLoader",
+    "org.apache.cassandra.distributed.impl.InstanceClassLoader",
-    "org.apache.cassandra.distributed.InstanceConfig",
+    "org.apache.cassandra.distributed.impl.InstanceConfig",
-    "org.apache.cassandra.distributed.InvokableInstance",
+    "org.apache.cassandra.distributed.impl.InvokableInstance",
-    "org.apache.cassandra.distributed.InvokableInstance$CallableNoExcept",
+    "org.apache.cassandra.distributed.impl.InvokableInstance$CallableNoExcept",
-    "org.apache.cassandra.distributed.InvokableInstance$InstanceFunction",
+    "org.apache.cassandra.distributed.impl.InvokableInstance$InstanceFunction",
-    "org.apache.cassandra.distributed.InvokableInstance$SerializableBiConsumer",
+    "org.apache.cassandra.distributed.impl.InvokableInstance$SerializableBiConsumer",
-    "org.apache.cassandra.distributed.InvokableInstance$SerializableBiFunction",
+    "org.apache.cassandra.distributed.impl.InvokableInstance$SerializableBiFunction",
-    "org.apache.cassandra.distributed.InvokableInstance$SerializableCallable",
+    "org.apache.cassandra.distributed.impl.InvokableInstance$SerializableCallable",
-    "org.apache.cassandra.distributed.InvokableInstance$SerializableConsumer",
+    "org.apache.cassandra.distributed.impl.InvokableInstance$SerializableConsumer",
-    "org.apache.cassandra.distributed.InvokableInstance$SerializableFunction",
+    "org.apache.cassandra.distributed.impl.InvokableInstance$SerializableFunction",
-    "org.apache.cassandra.distributed.InvokableInstance$SerializableRunnable",
+    "org.apache.cassandra.distributed.impl.InvokableInstance$SerializableRunnable",
-    "org.apache.cassandra.distributed.InvokableInstance$SerializableTriFunction",
+    "org.apache.cassandra.distributed.impl.InvokableInstance$SerializableTriFunction",
-    "org.apache.cassandra.distributed.InvokableInstance$TriFunction",
+    "org.apache.cassandra.distributed.impl.InvokableInstance$TriFunction",
-    "org.apache.cassandra.distributed.Message",
+    "org.apache.cassandra.distributed.impl.Message",
     "org.apache.cassandra.exceptions.ConfigurationException",
     "org.apache.cassandra.exceptions.RequestValidationException",
     "org.apache.cassandra.exceptions.CassandraException",