[GitHub] [cassandra] jonmeredith commented on a change in pull request #780: Cannot replace_address /X because it doesn't exist in gossip

[GitBox <gi...@apache.org>]

jonmeredith commented on a change in pull request #780:
URL: https://github.com/apache/cassandra/pull/780#discussion_r535511321



##########
File path: src/java/org/apache/cassandra/gms/Gossiper.java
##########
@@ -1779,16 +1831,17 @@ public void addSavedEndpoint(InetAddressAndPort ep)
         if (epState != null)
         {
             logger.debug("not replacing a previous epState for {}, but reusing it: {}", ep, epState);
-            epState.setHeartBeatState(new HeartBeatState(0));
+            epState.setHeartBeatState(HeartBeatState.empty());
         }
         else
         {
-            epState = new EndpointState(new HeartBeatState(0));
+            epState = new EndpointState(HeartBeatState.empty());
+            logger.info("Adding {} as there was no previous epState; new state is {}", ep, epState);
         }
 
         epState.markDead();
         endpointStateMap.put(ep, epState);
-        unreachableEndpoints.put(ep, System.nanoTime());
+        markDead(ep, epState);

Review comment:
       This will now fire any subscribed listeners but more importantly diagnostics and log messages about the endpoint being DOWN that would not have fired before.
   
   This instance hasn't learned anything about the remote so not sure it's appropriate tell listeners/loggers that the node is dead and it's possible users have built/will build monitoring based on the diagnostic events that would be noisier as a result.
   
   Perhaps refactor the state changing parts into a `silentlyMarkDead()` method and just have `markDead()` call that instead.

##########
File path: src/java/org/apache/cassandra/gms/Gossiper.java
##########
@@ -1480,23 +1482,73 @@ private void sendAll(GossipDigest gDigest, Map<InetAddressAndPort, EndpointState
             deltaEpStateMap.put(gDigest.getEndpoint(), localEpStatePtr);
     }
 
-    /*
-        This method is used to figure the state that the Gossiper has but Gossipee doesn't. The delta digests
-        and the delta state are built up.
-    */
-    void examineGossiper(List<GossipDigest> gDigestList, List<GossipDigest> deltaGossipDigestList, Map<InetAddressAndPort, EndpointState> deltaEpStateMap)
+    /**
+     * Used during a shadow round to collect the current state; this method clones the current state, no filtering
+     * is done.
+     *
+     * Durning the shadow round its desirable to return gossip state for remote instances that were created by this

Review comment:
       Durning -> During

##########
File path: src/java/org/apache/cassandra/service/StorageService.java
##########
@@ -523,18 +523,68 @@ private synchronized UUID prepareForReplacement() throws ConfigurationException
         logger.info("Gathering node replacement information for {}", replaceAddress);
         Map<InetAddressAndPort, EndpointState> epStates = Gossiper.instance.doShadowRound();
         // as we've completed the shadow round of gossip, we should be able to find the node we're replacing
-        if (epStates.get(replaceAddress) == null)
+        EndpointState state = epStates.get(replaceAddress);
+        if (state == null)
             throw new RuntimeException(String.format("Cannot replace_address %s because it doesn't exist in gossip", replaceAddress));
 
         validateEndpointSnitch(epStates.values().iterator());
 
         try
         {
-            VersionedValue tokensVersionedValue = epStates.get(replaceAddress).getApplicationState(ApplicationState.TOKENS);
+            VersionedValue tokensVersionedValue = state.getApplicationState(ApplicationState.TOKENS);
             if (tokensVersionedValue == null)
                 throw new RuntimeException(String.format("Could not find tokens for %s to replace", replaceAddress));
 
-            bootstrapTokens = TokenSerializer.deserialize(tokenMetadata.partitioner, new DataInputStream(new ByteArrayInputStream(tokensVersionedValue.toBytes())));
+            Collection<Token> tokens = TokenSerializer.deserialize(tokenMetadata.partitioner, new DataInputStream(new ByteArrayInputStream(tokensVersionedValue.toBytes())));
+            bootstrapTokens = validateReplacementBootstrapTokens(tokenMetadata, replaceAddress, tokens);
+
+            if (state.isEmpty() && REPLACEMENT_ALLOW_EMPTY.getBoolean())
+            {
+                logger.warn("Gossip state is-empty for node {}. Injecting token metadata.", replaceAddress);

Review comment:
       maybe something targeted to operators more `Gossip state not present for replacing node {}. Adding temporary entry to mark dead.`
   

##########
File path: src/java/org/apache/cassandra/gms/Gossiper.java
##########
@@ -1895,6 +1948,26 @@ protected boolean isInShadowRound()
         return inShadowRound;
     }
 
+    /**
+     * Creates a new dead {@link EndpointState} that is {@link EndpointState#isEmpty() empty}.  This is used during
+     * host replacement for edge cases where the seed notified that the endpoint was empty, so need to add such state
+     * into gossip explicitly (as empty endpoints are not gossiped outside of the shadow round).
+     *
+     * see CASSANDRA-16213
+     */
+    public void initializeUnreachableNodeUnsafe(InetAddressAndPort addr)
+    {
+        EndpointState state = new EndpointState(HeartBeatState.empty());
+        state.markDead();
+        EndpointState oldState = endpointStateMap.putIfAbsent(addr, state);
+        if (null != oldState)
+        {
+            throw new RuntimeException("Attempted to initialize endpoint state for unreachable node, " +
+                                       "but found existing endpoint state for it. This should only " +
+                                       "happen during a replacement");

Review comment:
       Why include "This should only happen during a replacement"?
   
   The endpoint state shouldn't be updated as the gossiper hasn't started yet, and any pending shadow round replies from other nodes would not be included, so I don't think this should ever happen unless something else has been changed.

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)

Review comment:
       nit: src/target aren't obvious to me which way around, maybe `ringInstance` and `target` would be clearer (was originally going to suggest `expected` for target but I can see the other assert methods below where target makes more sense).

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");

Review comment:
       Should this use `--print-port` in 4.0 to support tests using `org.apache.cassandra.distributed.impl.INodeProvisionStrategy.Strategy#OneNetworkInterface`

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance's gossip state matches on the source instance
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @param state expected gossip state
+     * @return the ring (if target is present and has expected state)
+     */
+    public static List<RingInstanceDetails> assertRingState(IInstance src, IInstance target, String state)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        List<RingInstanceDetails> match = ring.stream()
+                                              .filter(d -> d.address.equals(targetAddress))

Review comment:
       Maybe port check if you decide to support it.

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance's gossip state matches on the source instance
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @param state expected gossip state
+     * @return the ring (if target is present and has expected state)
+     */
+    public static List<RingInstanceDetails> assertRingState(IInstance src, IInstance target, String state)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        List<RingInstanceDetails> match = ring.stream()
+                                              .filter(d -> d.address.equals(targetAddress))
+                                              .collect(Collectors.toList());
+        assertThat(match)
+        .isNotEmpty()
+        .as("State was expected to be %s but was not", state)

Review comment:
       `"...but was not present"`?  Might help to include the target ip/port in there too to disambiguate error messages. 

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance's gossip state matches on the source instance
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @param state expected gossip state
+     * @return the ring (if target is present and has expected state)
+     */
+    public static List<RingInstanceDetails> assertRingState(IInstance src, IInstance target, String state)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        List<RingInstanceDetails> match = ring.stream()
+                                              .filter(d -> d.address.equals(targetAddress))
+                                              .collect(Collectors.toList());
+        assertThat(match)
+        .isNotEmpty()
+        .as("State was expected to be %s but was not", state)
+        .allMatch(r -> r.state.equals(state));
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance is NOT in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance not expected in the ring
+     * @return the ring (if target is not present)
+     */
+    public static List<RingInstanceDetails> assertNotInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();

Review comment:
       Maybe port check if you decide to support it.

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance's gossip state matches on the source instance
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @param state expected gossip state
+     * @return the ring (if target is present and has expected state)
+     */
+    public static List<RingInstanceDetails> assertRingState(IInstance src, IInstance target, String state)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        List<RingInstanceDetails> match = ring.stream()
+                                              .filter(d -> d.address.equals(targetAddress))
+                                              .collect(Collectors.toList());
+        assertThat(match)
+        .isNotEmpty()
+        .as("State was expected to be %s but was not", state)
+        .allMatch(r -> r.state.equals(state));
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance is NOT in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance not expected in the ring
+     * @return the ring (if target is not present)
+     */
+    public static List<RingInstanceDetails> assertNotInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        Assert.assertEquals("Not expected to find " + targetAddress + " but was found", Optional.empty(), match);
+        return ring;
+    }
+
+    private static Map<String, Map<String, String>> awaitGossip(IInstance src, String errorMessage, Predicate<Map<String, Map<String, String>>> fn)
+    {
+        Map<String, Map<String, String>> gossip = null;
+        for (int i = 0; i < 100; i++)

Review comment:
       Fixed timeout might not suit all consumers, but could fix later if needed. 100s feels like a long time, but that depends what you're waiting for I suppose.

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance's gossip state matches on the source instance
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @param state expected gossip state
+     * @return the ring (if target is present and has expected state)
+     */
+    public static List<RingInstanceDetails> assertRingState(IInstance src, IInstance target, String state)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        List<RingInstanceDetails> match = ring.stream()
+                                              .filter(d -> d.address.equals(targetAddress))
+                                              .collect(Collectors.toList());
+        assertThat(match)
+        .isNotEmpty()
+        .as("State was expected to be %s but was not", state)
+        .allMatch(r -> r.state.equals(state));
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance is NOT in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance not expected in the ring
+     * @return the ring (if target is not present)
+     */
+    public static List<RingInstanceDetails> assertNotInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        Assert.assertEquals("Not expected to find " + targetAddress + " but was found", Optional.empty(), match);
+        return ring;
+    }
+
+    private static Map<String, Map<String, String>> awaitGossip(IInstance src, String errorMessage, Predicate<Map<String, Map<String, String>>> fn)
+    {
+        Map<String, Map<String, String>> gossip = null;
+        for (int i = 0; i < 100; i++)
+        {
+            gossip = gossipInfo(src);
+            if (fn.test(gossip))
+            {
+                return gossip;
+            }
+            sleepUninterruptibly(1, TimeUnit.SECONDS);
+        }
+        throw new AssertionError(errorMessage + "\n" + gossip);
+    }
+
+    /**
+     * Wait for the target instance to have the desired status. Target status is checked via string contains so works
+     * with 'NORMAL' but also can check tokens or full state.
+     *
+     * @param src instance to check on
+     * @param target instance to wait for
+     * @param targetStatus for the instance
+     * @return gossip info
+     */
+    public static Map<String, Map<String, String>> awaitGossipStatus(IInstance src, IInstance target, String targetStatus)
+    {
+        return awaitGossip(src, "Node " + target + " did not match state " + targetStatus, gossip -> {
+            Map<String, String> state = gossip.get(target.config().broadcastAddress().getAddress().toString());
+            if (state == null)
+                return false;
+            String status = state.get("STATUS_WITH_PORT");
+            if (status == null)
+                status = state.get("STATUS");
+            if (status == null)
+                return targetStatus == null;
+            return status.contains(targetStatus);
+        });
+    }
+
+    private static List<RingInstanceDetails> awaitRing(IInstance src, String errorMessage, Predicate<List<RingInstanceDetails>> fn)
+    {
+        List<RingInstanceDetails> ring = null;
+        for (int i = 0; i < 100; i++)
+        {
+            ring = ring(src);
+            if (fn.test(ring))
+            {
+                return ring;
+            }
+            sleepUninterruptibly(1, TimeUnit.SECONDS);
+        }
+        throw new AssertionError(errorMessage + "\n" + ring);
+    }
+
+    /**
+     * Wait for the target to be in the ring as seen by the source instance.
+     *
+     * @param src instance to check on
+     * @param target instance to wait for
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingJoin(IInstance src, IInstance target)
+    {
+        return awaitRingJoin(src, target.broadcastAddress().getAddress().getHostAddress());
+    }
+
+    /**
+     * Wait for the target to be in the ring as seen by the source instance.
+     *
+     * @param src instance to check on
+     * @param targetAddress instance address to wait for
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingJoin(IInstance src, String targetAddress)
+    {
+        return awaitRing(src, "Node " + targetAddress + " did not join the ring...", ring -> {
+            Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();

Review comment:
       Maybe port check if you decide to support it.

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance's gossip state matches on the source instance
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @param state expected gossip state
+     * @return the ring (if target is present and has expected state)
+     */
+    public static List<RingInstanceDetails> assertRingState(IInstance src, IInstance target, String state)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        List<RingInstanceDetails> match = ring.stream()
+                                              .filter(d -> d.address.equals(targetAddress))
+                                              .collect(Collectors.toList());
+        assertThat(match)
+        .isNotEmpty()
+        .as("State was expected to be %s but was not", state)
+        .allMatch(r -> r.state.equals(state));
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance is NOT in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance not expected in the ring
+     * @return the ring (if target is not present)
+     */
+    public static List<RingInstanceDetails> assertNotInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        Assert.assertEquals("Not expected to find " + targetAddress + " but was found", Optional.empty(), match);
+        return ring;
+    }
+
+    private static Map<String, Map<String, String>> awaitGossip(IInstance src, String errorMessage, Predicate<Map<String, Map<String, String>>> fn)
+    {
+        Map<String, Map<String, String>> gossip = null;
+        for (int i = 0; i < 100; i++)
+        {
+            gossip = gossipInfo(src);
+            if (fn.test(gossip))
+            {
+                return gossip;
+            }
+            sleepUninterruptibly(1, TimeUnit.SECONDS);
+        }
+        throw new AssertionError(errorMessage + "\n" + gossip);
+    }
+
+    /**
+     * Wait for the target instance to have the desired status. Target status is checked via string contains so works
+     * with 'NORMAL' but also can check tokens or full state.
+     *
+     * @param src instance to check on
+     * @param target instance to wait for
+     * @param targetStatus for the instance
+     * @return gossip info
+     */
+    public static Map<String, Map<String, String>> awaitGossipStatus(IInstance src, IInstance target, String targetStatus)
+    {
+        return awaitGossip(src, "Node " + target + " did not match state " + targetStatus, gossip -> {
+            Map<String, String> state = gossip.get(target.config().broadcastAddress().getAddress().toString());
+            if (state == null)
+                return false;
+            String status = state.get("STATUS_WITH_PORT");
+            if (status == null)
+                status = state.get("STATUS");
+            if (status == null)
+                return targetStatus == null;
+            return status.contains(targetStatus);
+        });
+    }
+
+    private static List<RingInstanceDetails> awaitRing(IInstance src, String errorMessage, Predicate<List<RingInstanceDetails>> fn)
+    {
+        List<RingInstanceDetails> ring = null;
+        for (int i = 0; i < 100; i++)
+        {
+            ring = ring(src);
+            if (fn.test(ring))
+            {
+                return ring;
+            }
+            sleepUninterruptibly(1, TimeUnit.SECONDS);
+        }
+        throw new AssertionError(errorMessage + "\n" + ring);
+    }
+
+    /**
+     * Wait for the target to be in the ring as seen by the source instance.
+     *
+     * @param src instance to check on
+     * @param target instance to wait for
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingJoin(IInstance src, IInstance target)
+    {
+        return awaitRingJoin(src, target.broadcastAddress().getAddress().getHostAddress());
+    }
+
+    /**
+     * Wait for the target to be in the ring as seen by the source instance.
+     *
+     * @param src instance to check on
+     * @param targetAddress instance address to wait for
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingJoin(IInstance src, String targetAddress)
+    {
+        return awaitRing(src, "Node " + targetAddress + " did not join the ring...", ring -> {
+            Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+            if (match.isPresent())
+            {
+                RingInstanceDetails details = match.get();
+                return details.status.equals("Up") && details.state.equals("Normal");
+            }
+            return false;
+        });
+    }
+
+    /**
+     * Wait for the ring to only have instances that are Up and Normal.
+     *
+     * @param src instance to check on
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingHealthy(IInstance src)
+    {
+        return awaitRing(src, "Timeout waiting for ring to become healthy",
+                         ring ->
+                         ring.stream().allMatch(ClusterUtils::isRingInstanceDetailsHealthy));
+    }
+
+    /**
+     * Wait for the ring to have the target instance with the provided state.
+     *
+     * @param src instance to check on
+     * @param target to look for
+     * @param state expected
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingState(IInstance src, IInstance target, String state)
+    {
+        return awaitRing(src, "Timeout waiting for " + target + " to have state " + state,
+                         ring ->
+                         ring.stream()
+                             .filter(d -> d.address.equals(target.config().broadcastAddress().getAddress().getHostAddress()))
+                             .filter(d -> d.state.equals(state))
+                             .findAny().isPresent());
+    }
+
+    /**
+     * Make sure the ring is only the expected instances.  The source instance may not be in the ring, so this function
+     * only relies on the expectedInsts param.
+     *
+     * @param src instance to check on
+     * @param expectedInsts expected instances in the ring
+     * @return the ring (if condition is true)
+     */
+    public static List<RingInstanceDetails> assertRingIs(IInstance src, IInstance... expectedInsts)
+    {
+        return assertRingIs(src, Arrays.asList(expectedInsts));
+    }
+
+    /**
+     * Make sure the ring is only the expected instances.  The source instance may not be in the ring, so this function
+     * only relies on the expectedInsts param.
+     *
+     * @param src instance to check on
+     * @param expectedInsts expected instances in the ring
+     * @return the ring (if condition is true)
+     */
+    public static List<RingInstanceDetails> assertRingIs(IInstance src, List<? extends IInstance> expectedInsts)
+    {
+        Set<String> expectedRingAddresses = expectedInsts.stream()
+                                                  .map(i -> i.config().broadcastAddress().getAddress().getHostAddress())

Review comment:
       Maybe include port if you decide to support it.

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/WithProperties.java
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.shared;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import com.google.common.base.Joiner;
+
+import org.apache.cassandra.config.CassandraRelevantProperties;
+
+public final class WithProperties implements AutoCloseable
+{
+    private final List<State> properties = new ArrayList<>();
+
+    public WithProperties()
+    {
+    }
+
+    public WithProperties(String... kvs)
+    {
+        with(kvs);
+    }
+
+    public void with(String... kvs)
+    {
+        assert kvs.length % 2 == 0 : "Input must have an even amount of inputs but given " + kvs.length;
+        for (int i = 0; i <= kvs.length - 2; i = i + 2)
+        {
+            with(kvs[i], kvs[i + 1]);
+        }
+    }
+
+    public void setProperty(String key, String value)
+    {
+        with(key, value);
+    }
+
+    public void set(CassandraRelevantProperties prop, String value)
+    {
+        with(prop.getKey(), value);
+    }
+
+    public void set(CassandraRelevantProperties prop, String... values)
+    {
+        set(prop, Arrays.asList(values));
+    }
+
+    public void set(CassandraRelevantProperties prop, Collection<String> values)
+    {
+        set(prop, Joiner.on(",").join(values));
+    }
+
+    public void set(CassandraRelevantProperties prop, boolean value)
+    {
+        set(prop, Boolean.toString(value));
+    }
+
+    public void set(CassandraRelevantProperties prop, long value)
+    {
+        set(prop, Long.toString(value));
+    }
+
+    public void with(String key, String value)
+    {
+        String previous = System.setProperty(key, value);
+        properties.add(new State(key, previous));
+    }
+
+
+    @Override
+    public void close()
+    {
+        Collections.reverse(properties);
+        properties.forEach(s -> {
+            if (s.value == null)
+                System.getProperties().remove(s.key);
+            else
+                System.setProperty(s.key, s.value);
+        });
+        properties.clear();
+    }
+
+    private static final class State

Review comment:
       Why `State` not `Property` and `key` not `name`?

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance's gossip state matches on the source instance
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @param state expected gossip state
+     * @return the ring (if target is present and has expected state)
+     */
+    public static List<RingInstanceDetails> assertRingState(IInstance src, IInstance target, String state)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        List<RingInstanceDetails> match = ring.stream()
+                                              .filter(d -> d.address.equals(targetAddress))
+                                              .collect(Collectors.toList());
+        assertThat(match)
+        .isNotEmpty()
+        .as("State was expected to be %s but was not", state)
+        .allMatch(r -> r.state.equals(state));
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance is NOT in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance not expected in the ring
+     * @return the ring (if target is not present)
+     */
+    public static List<RingInstanceDetails> assertNotInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        Assert.assertEquals("Not expected to find " + targetAddress + " but was found", Optional.empty(), match);
+        return ring;
+    }
+
+    private static Map<String, Map<String, String>> awaitGossip(IInstance src, String errorMessage, Predicate<Map<String, Map<String, String>>> fn)
+    {
+        Map<String, Map<String, String>> gossip = null;
+        for (int i = 0; i < 100; i++)
+        {
+            gossip = gossipInfo(src);
+            if (fn.test(gossip))
+            {
+                return gossip;
+            }
+            sleepUninterruptibly(1, TimeUnit.SECONDS);
+        }
+        throw new AssertionError(errorMessage + "\n" + gossip);
+    }
+
+    /**
+     * Wait for the target instance to have the desired status. Target status is checked via string contains so works
+     * with 'NORMAL' but also can check tokens or full state.
+     *
+     * @param src instance to check on
+     * @param target instance to wait for
+     * @param targetStatus for the instance
+     * @return gossip info
+     */
+    public static Map<String, Map<String, String>> awaitGossipStatus(IInstance src, IInstance target, String targetStatus)
+    {
+        return awaitGossip(src, "Node " + target + " did not match state " + targetStatus, gossip -> {
+            Map<String, String> state = gossip.get(target.config().broadcastAddress().getAddress().toString());
+            if (state == null)
+                return false;
+            String status = state.get("STATUS_WITH_PORT");
+            if (status == null)
+                status = state.get("STATUS");
+            if (status == null)
+                return targetStatus == null;
+            return status.contains(targetStatus);
+        });
+    }
+
+    private static List<RingInstanceDetails> awaitRing(IInstance src, String errorMessage, Predicate<List<RingInstanceDetails>> fn)
+    {
+        List<RingInstanceDetails> ring = null;
+        for (int i = 0; i < 100; i++)
+        {
+            ring = ring(src);
+            if (fn.test(ring))
+            {
+                return ring;
+            }
+            sleepUninterruptibly(1, TimeUnit.SECONDS);
+        }
+        throw new AssertionError(errorMessage + "\n" + ring);
+    }
+
+    /**
+     * Wait for the target to be in the ring as seen by the source instance.
+     *
+     * @param src instance to check on
+     * @param target instance to wait for
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingJoin(IInstance src, IInstance target)
+    {
+        return awaitRingJoin(src, target.broadcastAddress().getAddress().getHostAddress());
+    }
+
+    /**
+     * Wait for the target to be in the ring as seen by the source instance.
+     *
+     * @param src instance to check on
+     * @param targetAddress instance address to wait for
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingJoin(IInstance src, String targetAddress)
+    {
+        return awaitRing(src, "Node " + targetAddress + " did not join the ring...", ring -> {
+            Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+            if (match.isPresent())
+            {
+                RingInstanceDetails details = match.get();
+                return details.status.equals("Up") && details.state.equals("Normal");
+            }
+            return false;
+        });
+    }
+
+    /**
+     * Wait for the ring to only have instances that are Up and Normal.
+     *
+     * @param src instance to check on
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingHealthy(IInstance src)
+    {
+        return awaitRing(src, "Timeout waiting for ring to become healthy",
+                         ring ->
+                         ring.stream().allMatch(ClusterUtils::isRingInstanceDetailsHealthy));
+    }
+
+    /**
+     * Wait for the ring to have the target instance with the provided state.
+     *
+     * @param src instance to check on
+     * @param target to look for
+     * @param state expected
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingState(IInstance src, IInstance target, String state)
+    {
+        return awaitRing(src, "Timeout waiting for " + target + " to have state " + state,
+                         ring ->
+                         ring.stream()
+                             .filter(d -> d.address.equals(target.config().broadcastAddress().getAddress().getHostAddress()))

Review comment:
       Maybe port check if you decide to support it.

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance's gossip state matches on the source instance
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @param state expected gossip state
+     * @return the ring (if target is present and has expected state)
+     */
+    public static List<RingInstanceDetails> assertRingState(IInstance src, IInstance target, String state)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        List<RingInstanceDetails> match = ring.stream()
+                                              .filter(d -> d.address.equals(targetAddress))
+                                              .collect(Collectors.toList());
+        assertThat(match)
+        .isNotEmpty()
+        .as("State was expected to be %s but was not", state)
+        .allMatch(r -> r.state.equals(state));
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance is NOT in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance not expected in the ring
+     * @return the ring (if target is not present)
+     */
+    public static List<RingInstanceDetails> assertNotInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        Assert.assertEquals("Not expected to find " + targetAddress + " but was found", Optional.empty(), match);
+        return ring;
+    }
+
+    private static Map<String, Map<String, String>> awaitGossip(IInstance src, String errorMessage, Predicate<Map<String, Map<String, String>>> fn)
+    {
+        Map<String, Map<String, String>> gossip = null;
+        for (int i = 0; i < 100; i++)
+        {
+            gossip = gossipInfo(src);
+            if (fn.test(gossip))
+            {
+                return gossip;
+            }
+            sleepUninterruptibly(1, TimeUnit.SECONDS);
+        }
+        throw new AssertionError(errorMessage + "\n" + gossip);
+    }
+
+    /**
+     * Wait for the target instance to have the desired status. Target status is checked via string contains so works
+     * with 'NORMAL' but also can check tokens or full state.
+     *
+     * @param src instance to check on
+     * @param target instance to wait for
+     * @param targetStatus for the instance
+     * @return gossip info
+     */
+    public static Map<String, Map<String, String>> awaitGossipStatus(IInstance src, IInstance target, String targetStatus)
+    {
+        return awaitGossip(src, "Node " + target + " did not match state " + targetStatus, gossip -> {
+            Map<String, String> state = gossip.get(target.config().broadcastAddress().getAddress().toString());
+            if (state == null)
+                return false;
+            String status = state.get("STATUS_WITH_PORT");
+            if (status == null)
+                status = state.get("STATUS");
+            if (status == null)
+                return targetStatus == null;
+            return status.contains(targetStatus);
+        });
+    }
+
+    private static List<RingInstanceDetails> awaitRing(IInstance src, String errorMessage, Predicate<List<RingInstanceDetails>> fn)
+    {
+        List<RingInstanceDetails> ring = null;
+        for (int i = 0; i < 100; i++)
+        {
+            ring = ring(src);
+            if (fn.test(ring))
+            {
+                return ring;
+            }
+            sleepUninterruptibly(1, TimeUnit.SECONDS);
+        }
+        throw new AssertionError(errorMessage + "\n" + ring);
+    }
+
+    /**
+     * Wait for the target to be in the ring as seen by the source instance.
+     *
+     * @param src instance to check on
+     * @param target instance to wait for
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingJoin(IInstance src, IInstance target)
+    {
+        return awaitRingJoin(src, target.broadcastAddress().getAddress().getHostAddress());
+    }
+
+    /**
+     * Wait for the target to be in the ring as seen by the source instance.
+     *
+     * @param src instance to check on
+     * @param targetAddress instance address to wait for
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingJoin(IInstance src, String targetAddress)
+    {
+        return awaitRing(src, "Node " + targetAddress + " did not join the ring...", ring -> {
+            Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+            if (match.isPresent())
+            {
+                RingInstanceDetails details = match.get();
+                return details.status.equals("Up") && details.state.equals("Normal");
+            }
+            return false;
+        });
+    }
+
+    /**
+     * Wait for the ring to only have instances that are Up and Normal.
+     *
+     * @param src instance to check on
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingHealthy(IInstance src)
+    {
+        return awaitRing(src, "Timeout waiting for ring to become healthy",
+                         ring ->
+                         ring.stream().allMatch(ClusterUtils::isRingInstanceDetailsHealthy));
+    }
+
+    /**
+     * Wait for the ring to have the target instance with the provided state.
+     *
+     * @param src instance to check on
+     * @param target to look for
+     * @param state expected
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingState(IInstance src, IInstance target, String state)
+    {
+        return awaitRing(src, "Timeout waiting for " + target + " to have state " + state,
+                         ring ->
+                         ring.stream()
+                             .filter(d -> d.address.equals(target.config().broadcastAddress().getAddress().getHostAddress()))
+                             .filter(d -> d.state.equals(state))
+                             .findAny().isPresent());
+    }
+
+    /**
+     * Make sure the ring is only the expected instances.  The source instance may not be in the ring, so this function
+     * only relies on the expectedInsts param.
+     *
+     * @param src instance to check on
+     * @param expectedInsts expected instances in the ring
+     * @return the ring (if condition is true)
+     */
+    public static List<RingInstanceDetails> assertRingIs(IInstance src, IInstance... expectedInsts)
+    {
+        return assertRingIs(src, Arrays.asList(expectedInsts));
+    }
+
+    /**
+     * Make sure the ring is only the expected instances.  The source instance may not be in the ring, so this function
+     * only relies on the expectedInsts param.
+     *
+     * @param src instance to check on
+     * @param expectedInsts expected instances in the ring
+     * @return the ring (if condition is true)
+     */
+    public static List<RingInstanceDetails> assertRingIs(IInstance src, List<? extends IInstance> expectedInsts)
+    {
+        Set<String> expectedRingAddresses = expectedInsts.stream()
+                                                  .map(i -> i.config().broadcastAddress().getAddress().getHostAddress())
+                                                  .collect(Collectors.toSet());
+        return assertRingIs(src, expectedRingAddresses);
+    }
+
+    /**
+     * Make sure the ring is only the expected instances.  The source instance may not be in the ring, so this function
+     * only relies on the expectedInsts param.
+     *
+     * @param src instance to check on
+     * @param expectedRingAddresses expected instances addresses in the ring
+     * @return the ring (if condition is true)
+     */
+    public static List<RingInstanceDetails> assertRingIs(IInstance src, Set<String> expectedRingAddresses)
+    {
+        List<RingInstanceDetails> ring = ring(src);
+        Set<String> ringAddresses = ring.stream().map(d -> d.address).collect(Collectors.toSet());
+        assertThat(ringAddresses)
+                  .as("Ring addreses did not match for instance %s", src)
+                  .isEqualTo(expectedRingAddresses);
+        return ring;
+    }
+
+    private static boolean isRingInstanceDetailsHealthy(RingInstanceDetails details)
+    {
+        return details.status.equals("Up") && details.state.equals("Normal");
+    }
+
+    private static List<RingInstanceDetails> parseRing(String str)
+    {
+        // 127.0.0.3  rack0       Up     Normal  46.21 KB        100.00%             -1
+        Pattern pattern = Pattern.compile("^([0-9.]+)\\s+\\w+\\s+(\\w+)\\s+(\\w+).*?(-?\\d+)\\s*$");
+        List<RingInstanceDetails> details = new ArrayList<>();
+        String[] lines = str.split("\n");
+        for (String line : lines)
+        {
+            Matcher matcher = pattern.matcher(line);
+            if (!matcher.find())
+            {
+                continue;
+            }
+            details.add(new RingInstanceDetails(matcher.group(1), matcher.group(2), matcher.group(3), matcher.group(4)));
+        }
+
+        return details;
+    }
+
+    /**
+     * Get the gossip information from the node.  Currently only address, generation, and heartbeat are returned
+     *
+     * @param inst to check on
+     * @return gossip info
+     */
+    public static Map<String, Map<String, String>> gossipInfo(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("gossipinfo");
+        results.asserts().success();
+        return parseGossipInfo(results.getStdout());
+    }
+
+    /**
+     * Make sure the gossip info for the specific target has the expected generation and heartbeat
+     *
+     * @param inst to check on
+     * @param target instance to check for
+     * @param expectedGeneration expected generation
+     * @param expectedHeartbeat expected heartbeat
+     */
+    public static void assertGossipInfo(IInstance inst,
+                                        InetSocketAddress target, int expectedGeneration, int expectedHeartbeat)
+    {
+        String targetAddress = target.getAddress().toString();
+        Map<String, Map<String, String>> gossipInfo = gossipInfo(inst);
+        Map<String, String> gossipState = gossipInfo.get(targetAddress);
+        if (gossipState == null)
+            throw new NullPointerException("Unable to find gossip info for " + targetAddress + "; gossip info = " + gossipInfo);
+        Assert.assertEquals(Long.toString(expectedGeneration), gossipState.get("generation"));
+        Assert.assertEquals(Long.toString(expectedHeartbeat), gossipState.get("heartbeat")); //TODO do we really mix these two?
+    }
+
+    private static Map<String, Map<String, String>> parseGossipInfo(String str)
+    {
+        Map<String, Map<String, String>> map = new HashMap<>();
+        String[] lines = str.split("\n");
+        String currentInstance = null;
+        for (String line : lines)
+        {
+            if (line.startsWith("/"))
+            {
+                // start of new instance
+                currentInstance = line;
+                continue;
+            }
+            Objects.requireNonNull(currentInstance);
+            String[] kv = line.trim().split(":", 2);
+            assert kv.length == 2 : "When splitting line " + line + " expected 2 parts but not true";
+            Map<String, String> state = map.computeIfAbsent(currentInstance, ignore -> new HashMap<>());
+            state.put(kv[0], kv[1]);
+        }
+
+        return map;
+    }
+
+    /**
+     * Get the tokens assigned to the instance via config.  This method does not work if the instance has learned
+     * or generated its tokens.
+     *
+     * @param instance to get tokens from
+     * @return non-empty list of tokens
+     */
+    public static List<String> getTokens(IInstance instance)

Review comment:
       nit: rename `getTokensFromConfig`?

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));

Review comment:
       This should be fine now Alex updated the DistributedTestSnitch to read from the gossiper state.

##########
File path: test/distributed/org/apache/cassandra/distributed/test/hostreplacement/BaseAssassinatedCase.java
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.hostreplacement;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.assertRingState;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.awaitGossipStatus;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.getTokens;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.replaceHostAndStart;
+import static org.apache.cassandra.distributed.test.hostreplacement.HostReplacementTest.setupCluster;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public abstract class BaseAssassinatedCase extends TestBaseImpl
+{
+    abstract void consume(Cluster cluster, IInvokableInstance nodeToRemove);
+
+    protected String expectedMessage(IInvokableInstance nodeToRemove)
+    {
+        return "Cannot replace token " + getTokens(nodeToRemove).get(0) + " which does not exist!";
+    }
+
+    @Test
+    public void test() throws IOException
+    {
+        TokenSupplier even = TokenSupplier.evenlyDistributedTokens(3);
+        try (Cluster cluster = Cluster.build(3)
+                                      .withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK))
+                                      .withTokenSupplier(node -> even.token(node == 4 || node == 5 ? 2 : node))
+                                      .start())
+        {
+            IInvokableInstance seed = cluster.get(1);
+            IInvokableInstance nodeToRemove = cluster.get(2);
+            IInvokableInstance peer = cluster.get(3);
+
+            setupCluster(cluster);
+
+            consume(cluster, nodeToRemove);
+
+            assertRingState(seed, nodeToRemove, "Normal");
+
+            // assassinate the node
+            peer.nodetoolResult("assassinate", nodeToRemove.config().broadcastAddress().getAddress().getHostAddress())

Review comment:
       Also possible to test including a port number.

##########
File path: test/distributed/org/apache/cassandra/distributed/test/hostreplacement/AssassinatedEmptyNodeTest.java
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.hostreplacement;
+
+import java.net.InetSocketAddress;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+
+import static org.apache.cassandra.distributed.shared.ClusterUtils.assertGossipInfo;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.stopAll;
+
+/**
+ * If the operator attempts to assassinate the node before replacing it, this will cause the node to fail to start
+ * as the status is non-normal.
+ *
+ * The cluster is put into the "empty" state for the node to remove.
+ */
+public class AssassinatedEmptyNodeTest extends BaseAssassinatedCase
+{
+    // empty state does not include the token metadata, so when assassinate happens it will fail to find the token
+    @Override
+    protected String expectedMessage(IInvokableInstance nodeToRemove)
+    {
+        return "Could not find tokens for " + nodeToRemove.config().broadcastAddress() + " to replace";
+    }
+
+    @Override
+    void consume(Cluster cluster, IInvokableInstance nodeToRemove)
+    {
+        IInvokableInstance seed = cluster.get(1);

Review comment:
       replace magic numbers `1` and `3` with constant in `BaseAssasinatedCase` like `SEED_NUM`, `PEER_NUM`?

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance's gossip state matches on the source instance
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @param state expected gossip state
+     * @return the ring (if target is present and has expected state)
+     */
+    public static List<RingInstanceDetails> assertRingState(IInstance src, IInstance target, String state)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        List<RingInstanceDetails> match = ring.stream()
+                                              .filter(d -> d.address.equals(targetAddress))
+                                              .collect(Collectors.toList());
+        assertThat(match)
+        .isNotEmpty()
+        .as("State was expected to be %s but was not", state)
+        .allMatch(r -> r.state.equals(state));

Review comment:
       Why the allMatch? Should just be one target. Will this print the current and the expected state on failure?

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();

Review comment:
       Maybe add port check otherwise `OneNetworkInterface` might find other nodes?

##########
File path: test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
##########
@@ -0,0 +1,742 @@
+/*
+ * 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.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+    {
+        return start(inst, (ignore, prop) -> fn.accept(prop));
+    }
+
+    /**
+     * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+     */
+    public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+    {
+        try (WithProperties properties = new WithProperties())
+        {
+            fn.accept(inst, properties);
+            inst.startup();
+            return inst;
+        }
+    }
+
+    /**
+     * Stop an instance in a blocking manner.
+     *
+     * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+     * the exceptions and throw as runtime.
+     */
+    public static void stopUnchecked(IInstance i)
+    {
+        Futures.getUnchecked(i.shutdown());
+    }
+
+    /**
+     * Stops an instance abruptly.  This is done by blocking all messages to/from so all other instances are unable
+     * to communicate, then stopping the instance gracefully.
+     *
+     * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+     * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+     *
+     * @param cluster to filter messages to
+     * @param inst to shut down
+     */
+    public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+    {
+        // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+        IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+        IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+        try
+        {
+            stopUnchecked(inst);
+        }
+        finally
+        {
+            from.off();
+            to.off();
+        }
+    }
+
+    /**
+     * Stop all the instances in the cluster.  This function is differe than {@link ICluster#close()} as it doesn't
+     * clean up the cluster state, it only stops all the instances.
+     */
+    public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+    {
+        cluster.stream().forEach(ClusterUtils::stopUnchecked);
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack)
+    {
+        return addInstance(cluster, dc, rack, ignore -> {});
+    }
+
+    /**
+     * Create a new instance and add it to the cluster, without starting it.
+     *
+     * @param cluster to add to
+     * @param dc the instance should be in
+     * @param rack the instance should be in
+     * @param fn function to add to the config before starting
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+                                                      String dc, String rack,
+                                                      Consumer<IInstanceConfig> fn)
+    {
+        Objects.requireNonNull(dc, "dc");
+        Objects.requireNonNull(rack, "rack");
+
+        InstanceConfig config = cluster.newInstanceConfig();
+        //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+        // instances known about (at least to NetworkTopology and TokenStategy)
+        // this is very hidden, so should be more explicit
+        config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+        fn.accept(config);
+
+        return cluster.bootstrap(config);
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+    {
+        return replaceHostAndStart(cluster, toReplace, ignore -> {});
+    }
+
+    /**
+     * Create and start a new instance that replaces an existing instance.
+     *
+     * The instance will be in the same datacenter and rack as the existing instance.
+     *
+     * @param cluster to add to
+     * @param toReplace instance to replace
+     * @param fn lambda to add additional properties
+     * @param <I> instance type
+     * @return the instance added
+     */
+    public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+                                                              IInstance toReplace,
+                                                              Consumer<WithProperties> fn)
+    {
+        IInstanceConfig toReplaceConf = toReplace.config();
+        I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+        return start(inst, properties -> {
+            // lower this so the replacement waits less time
+            properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+            // default is 30s, lowering as it should be faster
+            properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+            properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+            // state which node to replace
+            properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+            fn.accept(properties);
+        });
+    }
+
+    /**
+     * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+     */
+    public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+    {
+        return inst.callOnInstance(() ->
+                                   StorageService.instance.getTokenMetadata()
+                                                          .sortedTokens().stream()
+                                                          .map(Object::toString)
+                                                          .collect(Collectors.toList()));
+    }
+
+    /**
+     * Get the ring from the perspective of the instance.
+     */
+    public static List<RingInstanceDetails> ring(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("ring");
+        results.asserts().success();
+        return parseRing(results.getStdout());
+    }
+
+    /**
+     * Make sure the target instance is in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @return the ring (if target is present)
+     */
+    public static List<RingInstanceDetails> assertInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance's gossip state matches on the source instance
+     *
+     * @param src instance to check on
+     * @param target instance expected in the ring
+     * @param state expected gossip state
+     * @return the ring (if target is present and has expected state)
+     */
+    public static List<RingInstanceDetails> assertRingState(IInstance src, IInstance target, String state)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        List<RingInstanceDetails> match = ring.stream()
+                                              .filter(d -> d.address.equals(targetAddress))
+                                              .collect(Collectors.toList());
+        assertThat(match)
+        .isNotEmpty()
+        .as("State was expected to be %s but was not", state)
+        .allMatch(r -> r.state.equals(state));
+        return ring;
+    }
+
+    /**
+     * Make sure the target instance is NOT in the ring.
+     *
+     * @param src instance to check on
+     * @param target instance not expected in the ring
+     * @return the ring (if target is not present)
+     */
+    public static List<RingInstanceDetails> assertNotInRing(IInstance src, IInstance target)
+    {
+        String targetAddress = target.config().broadcastAddress().getAddress().getHostAddress();
+        List<RingInstanceDetails> ring = ring(src);
+        Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+        Assert.assertEquals("Not expected to find " + targetAddress + " but was found", Optional.empty(), match);
+        return ring;
+    }
+
+    private static Map<String, Map<String, String>> awaitGossip(IInstance src, String errorMessage, Predicate<Map<String, Map<String, String>>> fn)
+    {
+        Map<String, Map<String, String>> gossip = null;
+        for (int i = 0; i < 100; i++)
+        {
+            gossip = gossipInfo(src);
+            if (fn.test(gossip))
+            {
+                return gossip;
+            }
+            sleepUninterruptibly(1, TimeUnit.SECONDS);
+        }
+        throw new AssertionError(errorMessage + "\n" + gossip);
+    }
+
+    /**
+     * Wait for the target instance to have the desired status. Target status is checked via string contains so works
+     * with 'NORMAL' but also can check tokens or full state.
+     *
+     * @param src instance to check on
+     * @param target instance to wait for
+     * @param targetStatus for the instance
+     * @return gossip info
+     */
+    public static Map<String, Map<String, String>> awaitGossipStatus(IInstance src, IInstance target, String targetStatus)
+    {
+        return awaitGossip(src, "Node " + target + " did not match state " + targetStatus, gossip -> {
+            Map<String, String> state = gossip.get(target.config().broadcastAddress().getAddress().toString());
+            if (state == null)
+                return false;
+            String status = state.get("STATUS_WITH_PORT");
+            if (status == null)
+                status = state.get("STATUS");
+            if (status == null)
+                return targetStatus == null;
+            return status.contains(targetStatus);
+        });
+    }
+
+    private static List<RingInstanceDetails> awaitRing(IInstance src, String errorMessage, Predicate<List<RingInstanceDetails>> fn)
+    {
+        List<RingInstanceDetails> ring = null;
+        for (int i = 0; i < 100; i++)
+        {
+            ring = ring(src);
+            if (fn.test(ring))
+            {
+                return ring;
+            }
+            sleepUninterruptibly(1, TimeUnit.SECONDS);
+        }
+        throw new AssertionError(errorMessage + "\n" + ring);
+    }
+
+    /**
+     * Wait for the target to be in the ring as seen by the source instance.
+     *
+     * @param src instance to check on
+     * @param target instance to wait for
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingJoin(IInstance src, IInstance target)
+    {
+        return awaitRingJoin(src, target.broadcastAddress().getAddress().getHostAddress());
+    }
+
+    /**
+     * Wait for the target to be in the ring as seen by the source instance.
+     *
+     * @param src instance to check on
+     * @param targetAddress instance address to wait for
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingJoin(IInstance src, String targetAddress)
+    {
+        return awaitRing(src, "Node " + targetAddress + " did not join the ring...", ring -> {
+            Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+            if (match.isPresent())
+            {
+                RingInstanceDetails details = match.get();
+                return details.status.equals("Up") && details.state.equals("Normal");
+            }
+            return false;
+        });
+    }
+
+    /**
+     * Wait for the ring to only have instances that are Up and Normal.
+     *
+     * @param src instance to check on
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingHealthy(IInstance src)
+    {
+        return awaitRing(src, "Timeout waiting for ring to become healthy",
+                         ring ->
+                         ring.stream().allMatch(ClusterUtils::isRingInstanceDetailsHealthy));
+    }
+
+    /**
+     * Wait for the ring to have the target instance with the provided state.
+     *
+     * @param src instance to check on
+     * @param target to look for
+     * @param state expected
+     * @return the ring
+     */
+    public static List<RingInstanceDetails> awaitRingState(IInstance src, IInstance target, String state)
+    {
+        return awaitRing(src, "Timeout waiting for " + target + " to have state " + state,
+                         ring ->
+                         ring.stream()
+                             .filter(d -> d.address.equals(target.config().broadcastAddress().getAddress().getHostAddress()))
+                             .filter(d -> d.state.equals(state))
+                             .findAny().isPresent());
+    }
+
+    /**
+     * Make sure the ring is only the expected instances.  The source instance may not be in the ring, so this function
+     * only relies on the expectedInsts param.
+     *
+     * @param src instance to check on
+     * @param expectedInsts expected instances in the ring
+     * @return the ring (if condition is true)
+     */
+    public static List<RingInstanceDetails> assertRingIs(IInstance src, IInstance... expectedInsts)
+    {
+        return assertRingIs(src, Arrays.asList(expectedInsts));
+    }
+
+    /**
+     * Make sure the ring is only the expected instances.  The source instance may not be in the ring, so this function
+     * only relies on the expectedInsts param.
+     *
+     * @param src instance to check on
+     * @param expectedInsts expected instances in the ring
+     * @return the ring (if condition is true)
+     */
+    public static List<RingInstanceDetails> assertRingIs(IInstance src, List<? extends IInstance> expectedInsts)
+    {
+        Set<String> expectedRingAddresses = expectedInsts.stream()
+                                                  .map(i -> i.config().broadcastAddress().getAddress().getHostAddress())
+                                                  .collect(Collectors.toSet());
+        return assertRingIs(src, expectedRingAddresses);
+    }
+
+    /**
+     * Make sure the ring is only the expected instances.  The source instance may not be in the ring, so this function
+     * only relies on the expectedInsts param.
+     *
+     * @param src instance to check on
+     * @param expectedRingAddresses expected instances addresses in the ring
+     * @return the ring (if condition is true)
+     */
+    public static List<RingInstanceDetails> assertRingIs(IInstance src, Set<String> expectedRingAddresses)
+    {
+        List<RingInstanceDetails> ring = ring(src);
+        Set<String> ringAddresses = ring.stream().map(d -> d.address).collect(Collectors.toSet());
+        assertThat(ringAddresses)
+                  .as("Ring addreses did not match for instance %s", src)
+                  .isEqualTo(expectedRingAddresses);
+        return ring;
+    }
+
+    private static boolean isRingInstanceDetailsHealthy(RingInstanceDetails details)
+    {
+        return details.status.equals("Up") && details.state.equals("Normal");
+    }
+
+    private static List<RingInstanceDetails> parseRing(String str)
+    {
+        // 127.0.0.3  rack0       Up     Normal  46.21 KB        100.00%             -1
+        Pattern pattern = Pattern.compile("^([0-9.]+)\\s+\\w+\\s+(\\w+)\\s+(\\w+).*?(-?\\d+)\\s*$");
+        List<RingInstanceDetails> details = new ArrayList<>();
+        String[] lines = str.split("\n");
+        for (String line : lines)
+        {
+            Matcher matcher = pattern.matcher(line);
+            if (!matcher.find())
+            {
+                continue;
+            }
+            details.add(new RingInstanceDetails(matcher.group(1), matcher.group(2), matcher.group(3), matcher.group(4)));
+        }
+
+        return details;
+    }
+
+    /**
+     * Get the gossip information from the node.  Currently only address, generation, and heartbeat are returned
+     *
+     * @param inst to check on
+     * @return gossip info
+     */
+    public static Map<String, Map<String, String>> gossipInfo(IInstance inst)
+    {
+        NodeToolResult results = inst.nodetoolResult("gossipinfo");
+        results.asserts().success();
+        return parseGossipInfo(results.getStdout());
+    }
+
+    /**
+     * Make sure the gossip info for the specific target has the expected generation and heartbeat
+     *
+     * @param inst to check on
+     * @param target instance to check for
+     * @param expectedGeneration expected generation
+     * @param expectedHeartbeat expected heartbeat
+     */
+    public static void assertGossipInfo(IInstance inst,
+                                        InetSocketAddress target, int expectedGeneration, int expectedHeartbeat)
+    {
+        String targetAddress = target.getAddress().toString();
+        Map<String, Map<String, String>> gossipInfo = gossipInfo(inst);
+        Map<String, String> gossipState = gossipInfo.get(targetAddress);
+        if (gossipState == null)
+            throw new NullPointerException("Unable to find gossip info for " + targetAddress + "; gossip info = " + gossipInfo);
+        Assert.assertEquals(Long.toString(expectedGeneration), gossipState.get("generation"));
+        Assert.assertEquals(Long.toString(expectedHeartbeat), gossipState.get("heartbeat")); //TODO do we really mix these two?
+    }
+
+    private static Map<String, Map<String, String>> parseGossipInfo(String str)
+    {
+        Map<String, Map<String, String>> map = new HashMap<>();
+        String[] lines = str.split("\n");
+        String currentInstance = null;
+        for (String line : lines)
+        {
+            if (line.startsWith("/"))
+            {
+                // start of new instance
+                currentInstance = line;
+                continue;
+            }
+            Objects.requireNonNull(currentInstance);
+            String[] kv = line.trim().split(":", 2);
+            assert kv.length == 2 : "When splitting line " + line + " expected 2 parts but not true";

Review comment:
       nit: add quotes around 'line' when printing in case of unusual characters

##########
File path: test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementOfDowedClusterTest.java
##########
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.hostreplacement;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.List;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.SimpleQueryResult;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.assertj.core.api.Assertions;
+
+import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIPER_QUARANTINE_DELAY;
+import static org.apache.cassandra.config.CassandraRelevantProperties.REPLACEMENT_ALLOW_EMPTY;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.assertGossipInfo;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.assertNotInRing;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.assertRingIs;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.awaitRingHealthy;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.awaitRingJoin;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.getTokenMetadataTokens;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.replaceHostAndStart;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.stopAll;
+import static org.apache.cassandra.distributed.test.hostreplacement.HostReplacementTest.setupCluster;
+import static org.apache.cassandra.distributed.test.hostreplacement.HostReplacementTest.validateRows;
+
+public class HostReplacementOfDowedClusterTest extends TestBaseImpl
+{
+    private static final Logger logger = LoggerFactory.getLogger(HostReplacementOfDowedClusterTest.class);
+
+    static
+    {
+        // by default empty replace isn't allowed, so enable for these tests
+        REPLACEMENT_ALLOW_EMPTY.setBoolean(true);
+
+        // Gossip has a notiion of quarantine, which is used to remove "fat clients" and "gossip only members"

Review comment:
       `notiion` -> `notion`




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org