diff --git a/CHANGES.txt b/CHANGES.txt
index b4b1245881..ec88756517 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -43,6 +43,7 @@ Merged from 2.2:
  * Fixing invalid CQL in security documentation (CASSANDRA-15020)
  * Multi-version in-JVM dtests (CASSANDRA-14937)
  * Allow instance class loaders to be garbage collected for inJVM dtest (CASSANDRA-15170)
+ * Add support for network topology and query tracing for inJVM dtest (CASSANDRA-15319)
 
 
 3.11.4
diff --git a/build.xml b/build.xml
index 0b4e42327a..557cc076fe 100644
--- a/build.xml
+++ b/build.xml
@@ -1264,7 +1264,7 @@
              algorithm to limit the metaspace size and clean up SoftReferences
              more aggressively rather than waiting. See CASSANDRA-14922 for more details.
         -->
-        <jvmarg value="-XX:MaxMetaspaceSize=256M" />
+        <jvmarg value="-XX:MaxMetaspaceSize=384M" />
         <jvmarg value="-XX:SoftRefLRUPolicyMSPerMB=0" />
         <jvmarg value="-Dcassandra.memtable_row_overhead_computation_step=100"/>
         <jvmarg value="-Dcassandra.test.use_prepared=${cassandra.test.use_prepared}"/>
diff --git a/ide/idea/workspace.xml b/ide/idea/workspace.xml
index 8d1b0fce1e..7e63f77f8c 100644
--- a/ide/idea/workspace.xml
+++ b/ide/idea/workspace.xml
@@ -167,7 +167,7 @@
       <option name="MAIN_CLASS_NAME" value="" />
       <option name="METHOD_NAME" value="" />
       <option name="TEST_OBJECT" value="class" />
-      <option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -ea -XX:MaxMetaspaceSize=256M -XX:SoftRefLRUPolicyMSPerMB=0 -Dcassandra.strict.runtime.checks=true" />
+      <option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -ea -XX:MaxMetaspaceSize=384M -XX:SoftRefLRUPolicyMSPerMB=0 -Dcassandra.strict.runtime.checks=true" />
       <option name="PARAMETERS" value="" />
       <option name="WORKING_DIRECTORY" value="" />
       <option name="ENV_VARIABLES" />
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 309be3d26b..f34e5c2b3f 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -1039,7 +1039,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         }
 
         // if we don't have system_traces keyspace at this point, then create it manually
-        maybeAddOrUpdateKeyspace(TraceKeyspace.metadata());
+        ensureTraceKeyspace();
         maybeAddOrUpdateKeyspace(SystemDistributedKeyspace.metadata());
 
         if (!isSurveyMode)
@@ -1070,6 +1070,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         }
     }
 
+    @VisibleForTesting
+    public void ensureTraceKeyspace()
+    {
+        maybeAddOrUpdateKeyspace(TraceKeyspace.metadata());
+    }
+
     public static boolean isReplacingSameAddress()
     {
         InetAddress replaceAddress = DatabaseDescriptor.getReplaceAddress();
diff --git a/test/distributed/org/apache/cassandra/distributed/Cluster.java b/test/distributed/org/apache/cassandra/distributed/Cluster.java
index 95862b6a62..d3533d3fa9 100644
--- a/test/distributed/org/apache/cassandra/distributed/Cluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/Cluster.java
@@ -45,9 +45,14 @@ public class Cluster extends AbstractCluster<IInvokableInstance> implements IClu
         return new Wrapper(generation, version, config);
     }
 
+    public static Builder<IInvokableInstance, Cluster> build()
+    {
+        return new Builder<>(Cluster::new);
+    }
+
     public static Builder<IInvokableInstance, Cluster> build(int nodeCount)
     {
-        return new Builder<>(nodeCount, Cluster::new);
+        return build().withNodes(nodeCount);
     }
 
     public static Cluster create(int nodeCount, Consumer<InstanceConfig> configUpdater) throws IOException
diff --git a/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java b/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java
index c77d72514b..1fe960a253 100644
--- a/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java
@@ -46,9 +46,14 @@ public class UpgradeableCluster extends AbstractCluster<IUpgradeableInstance> im
         return new Wrapper(generation, version, config);
     }
 
+    public static Builder<IUpgradeableInstance, UpgradeableCluster> build()
+    {
+        return new Builder<>(UpgradeableCluster::new);
+    }
+
     public static Builder<IUpgradeableInstance, UpgradeableCluster> build(int nodeCount)
     {
-        return new Builder<>(nodeCount, UpgradeableCluster::new);
+        return build().withNodes(nodeCount);
     }
 
     public static UpgradeableCluster create(int nodeCount) throws Throwable
diff --git a/test/distributed/org/apache/cassandra/distributed/api/ICluster.java b/test/distributed/org/apache/cassandra/distributed/api/ICluster.java
index 91da61ed34..091e5f048a 100644
--- a/test/distributed/org/apache/cassandra/distributed/api/ICluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/api/ICluster.java
@@ -29,6 +29,8 @@ public interface ICluster
     IInstance get(InetAddressAndPort endpoint);
     int size();
     Stream<? extends IInstance> stream();
+    Stream<? extends IInstance> stream(String dcName);
+    Stream<? extends IInstance> stream(String dcName, String rackName);
     IMessageFilters filters();
 
 }
diff --git a/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java b/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java
index 59afd8bde6..ef4485318b 100644
--- a/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java
+++ b/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java
@@ -19,6 +19,8 @@
 package org.apache.cassandra.distributed.api;
 
 import java.util.Iterator;
+import java.util.UUID;
+import java.util.concurrent.Future;
 
 // The cross-version API requires that a Coordinator can be constructed without any constructor arguments
 public interface ICoordinator
@@ -28,4 +30,7 @@ public interface ICoordinator
     Object[][] execute(String query, Enum<?> consistencyLevel, Object... boundValues);
 
     Iterator<Object[]> executeWithPaging(String query, Enum<?> consistencyLevel, int pageSize, Object... boundValues);
+
+    Future<Object[][]> asyncExecuteWithTracing(UUID sessionId, String query, Enum<?> consistencyLevel, Object... boundValues);
+    Object[][] executeWithTracing(UUID sessionId, String query, Enum<?> consistencyLevel, Object... boundValues);
 }
diff --git a/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java
index 3e5a18fe7c..effa4296e9 100644
--- a/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java
+++ b/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java
@@ -18,8 +18,11 @@
 
 package org.apache.cassandra.distributed.api;
 
+import org.apache.cassandra.distributed.impl.NetworkTopology;
 import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.utils.Pair;
 
+import java.util.Map;
 import java.util.UUID;
 
 public interface IInstanceConfig
@@ -27,6 +30,17 @@ public interface IInstanceConfig
     int num();
     UUID hostId();
     InetAddressAndPort broadcastAddressAndPort();
+    NetworkTopology networkTopology();
+
+    default public String localRack()
+    {
+        return networkTopology().localRack(broadcastAddressAndPort());
+    }
+
+    default public String localDatacenter()
+    {
+        return networkTopology().localDC(broadcastAddressAndPort());
+    }
 
     /**
      * write the specified parameters to the Config object; we do not specify Config as the type to support a Config
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
index d09ed9d819..32006610d6 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
@@ -35,6 +35,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
 import com.google.common.collect.Sets;
@@ -57,6 +58,7 @@ import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.concurrent.SimpleCondition;
 
 /**
@@ -193,13 +195,15 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
         }
     }
 
-    protected AbstractCluster(File root, Versions.Version version, List<InstanceConfig> configs, ClassLoader sharedClassLoader)
+    protected AbstractCluster(File root, Versions.Version version, List<InstanceConfig> configs,
+                              ClassLoader sharedClassLoader)
     {
         this.root = root;
         this.sharedClassLoader = sharedClassLoader;
         this.instances = new ArrayList<>();
         this.instanceMap = new HashMap<>();
         int generation = AbstractCluster.generation.incrementAndGet();
+
         for (InstanceConfig config : configs)
         {
             I instance = newInstanceWrapper(generation, version, config);
@@ -231,7 +235,20 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
     {
         return instances.size();
     }
+
     public Stream<I> stream() { return instances.stream(); }
+
+    public Stream<I> stream(String dcName)
+    {
+        return instances.stream().filter(i -> i.config().localDatacenter().equals(dcName));
+    }
+
+    public Stream<I> stream(String dcName, String rackName)
+    {
+        return instances.stream().filter(i -> i.config().localDatacenter().equals(dcName) &&
+                                              i.config().localRack().equals(rackName));
+    }
+
     public void forEach(IIsolatedExecutor.SerializableRunnable runnable) { forEach(i -> i.sync(runnable)); }
     public void forEach(Consumer<? super I> consumer) { instances.forEach(consumer); }
     public void parallelForEach(IIsolatedExecutor.SerializableConsumer<? super I> consumer, long timeout, TimeUnit units)
@@ -309,7 +326,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
 
         private void signal()
         {
-            if (schemaHasChanged && 1 == instances.stream().map(IInstance::schemaVersion).distinct().count())
+            if (schemaHasChanged && 1 == instances.stream().filter(i -> !i.isShutdown()).map(IInstance::schemaVersion).distinct().count())
                 agreement.signalAll();
         }
 
@@ -353,15 +370,16 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
 
     public static class Builder<I extends IInstance, C extends AbstractCluster<I>>
     {
-        private final int nodeCount;
         private final Factory<I, C> factory;
+        private int nodeCount;
         private int subnet;
+        private Map<Integer, Pair<String,String>> nodeIdTopology;
         private File root;
         private Versions.Version version;
         private Consumer<InstanceConfig> configUpdater;
-        public Builder(int nodeCount, Factory<I, C> factory)
+
+        public Builder(Factory<I, C> factory)
         {
-            this.nodeCount = nodeCount;
             this.factory = factory;
         }
 
@@ -371,6 +389,97 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
             return this;
         }
 
+        public Builder<I, C> withNodes(int nodeCount) {
+            this.nodeCount = nodeCount;
+            return this;
+        }
+
+        public Builder<I, C> withDCs(int dcCount)
+        {
+            return withRacks(dcCount, 1);
+        }
+
+        public Builder<I, C> withRacks(int dcCount, int racksPerDC)
+        {
+            if (nodeCount == 0)
+                throw new IllegalStateException("Node count will be calculated. Do not supply total node count in the builder");
+
+            int totalRacks = dcCount * racksPerDC;
+            int nodesPerRack = (nodeCount + totalRacks - 1) / totalRacks; // round up to next integer
+            return withRacks(dcCount, racksPerDC, nodesPerRack);
+        }
+
+        public Builder<I, C> withRacks(int dcCount, int racksPerDC, int nodesPerRack)
+        {
+            if (nodeIdTopology != null)
+                throw new IllegalStateException("Network topology already created. Call withDCs/withRacks once or before withDC/withRack calls");
+
+            nodeIdTopology = new HashMap<>();
+            int nodeId = 1;
+            for (int dc = 1; dc <= dcCount; dc++)
+            {
+                for (int rack = 1; rack <= racksPerDC; rack++)
+                {
+                    for (int rackNodeIdx = 0; rackNodeIdx < nodesPerRack; rackNodeIdx++)
+                        nodeIdTopology.put(nodeId++, Pair.create(dcName(dc), rackName(rack)));
+                }
+            }
+            // adjust the node count to match the allocatation
+            final int adjustedNodeCount = dcCount * racksPerDC * nodesPerRack;
+            if (adjustedNodeCount != nodeCount)
+            {
+                assert adjustedNodeCount > nodeCount : "withRacks should only ever increase the node count";
+                logger.info("Network topology of {} DCs with {} racks per DC and {} nodes per rack required increasing total nodes to {}",
+                            dcCount, racksPerDC, nodesPerRack, adjustedNodeCount);
+                nodeCount = adjustedNodeCount;
+            }
+            return this;
+        }
+
+        public Builder<I, C> withDC(String dcName, int nodeCount)
+        {
+            return withRack(dcName, rackName(1), nodeCount);
+        }
+
+        public Builder<I, C> withRack(String dcName, String rackName, int nodesInRack)
+        {
+            if (nodeIdTopology == null)
+            {
+                if (nodeCount > 0)
+                    throw new IllegalStateException("Node count must not be explicitly set, or allocated using withDCs/withRacks");
+
+                nodeIdTopology = new HashMap<>();
+            }
+            for (int nodeId = nodeCount + 1; nodeId <= nodeCount + nodesInRack; nodeId++)
+                nodeIdTopology.put(nodeId, Pair.create(dcName, rackName));
+
+            nodeCount += nodesInRack;
+            return this;
+        }
+
+        // Map of node ids to dc and rack - must be contiguous with an entry nodeId 1 to nodeCount
+        public Builder<I, C> withNodeIdTopology(Map<Integer,Pair<String,String>> nodeIdTopology)
+        {
+            if (nodeIdTopology.isEmpty())
+                throw new IllegalStateException("Topology is empty. It must have an entry for every nodeId.");
+
+            IntStream.rangeClosed(1, nodeIdTopology.size()).forEach(nodeId -> {
+                if (nodeIdTopology.get(nodeId) == null)
+                    throw new IllegalStateException("Topology is missing entry for nodeId " + nodeId);
+            });
+
+            if (nodeCount != nodeIdTopology.size())
+            {
+                nodeCount = nodeIdTopology.size();
+                logger.info("Adjusting node count to {} for supplied network topology", nodeCount);
+
+            }
+
+            this.nodeIdTopology = new HashMap<>(nodeIdTopology);
+
+            return this;
+        }
+
         public Builder<I, C> withRoot(File root)
         {
             this.root = root;
@@ -396,9 +505,18 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
 
             if (root == null)
                 root = Files.createTempDirectory("dtests").toFile();
+
             if (version == null)
                 version = Versions.CURRENT;
 
+            if (nodeCount <= 0)
+                throw new IllegalStateException("Cluster must have at least one node");
+
+            if (nodeIdTopology == null)
+                nodeIdTopology = IntStream.rangeClosed(1, nodeCount).boxed()
+                                          .collect(Collectors.toMap(nodeId -> nodeId,
+                                                                    nodeId -> Pair.create(dcName(0), rackName(0))));
+
             root.mkdirs();
             setupLogging(root);
 
@@ -406,17 +524,23 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
 
             List<InstanceConfig> configs = new ArrayList<>();
             long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount);
-            for (int i = 0; i < nodeCount; ++i)
+
+            String ipPrefix = "127.0." + subnet + ".";
+
+            NetworkTopology networkTopology = NetworkTopology.build(ipPrefix, 7012, nodeIdTopology);
+
+            for (int i = 0 ; i < nodeCount ; ++i)
             {
-                InstanceConfig config = InstanceConfig.generate(i + 1, subnet, root, String.valueOf(token));
+                int nodeNum = i + 1;
+                String ipAddress = ipPrefix + nodeNum;
+                InstanceConfig config = InstanceConfig.generate(i + 1, ipAddress, networkTopology, root, String.valueOf(token));
                 if (configUpdater != null)
                     configUpdater.accept(config);
                 configs.add(config);
                 token += increment;
             }
 
-            C cluster = factory.newCluster(root, version, configs, sharedClassLoader);
-            return cluster;
+            return factory.newCluster(root, version, configs, sharedClassLoader);
         }
 
         public C start() throws IOException
@@ -427,6 +551,15 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
         }
     }
 
+    static String dcName(int index)
+    {
+        return "datacenter" + index;
+    }
+
+    static String rackName(int index)
+    {
+        return "rack" + index;
+    }
 
     private static void setupLogging(File root)
     {
@@ -434,11 +567,13 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
         {
             String testConfPath = "test/conf/logback-dtest.xml";
             Path logConfPath = Paths.get(root.getPath(), "/logback-dtest.xml");
+
             if (!logConfPath.toFile().exists())
             {
                 Files.copy(new File(testConfPath).toPath(),
                            logConfPath);
             }
+
             System.setProperty("logback.configurationFile", "file://" + logConfPath);
         }
         catch (IOException e)
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java
index ec5e58f8b4..97a3aa8d6e 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java
@@ -21,8 +21,11 @@ package org.apache.cassandra.distributed.impl;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.Future;
 
 import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.cql3.QueryOptions;
@@ -35,6 +38,7 @@ import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.QueryState;
 import org.apache.cassandra.service.pager.QueryPager;
 import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.transport.messages.ResultMessage;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
@@ -50,37 +54,56 @@ public class Coordinator implements ICoordinator
     @Override
     public Object[][] execute(String query, Enum<?> consistencyLevelOrigin, Object... boundValues)
     {
-        return instance.sync(() -> {
-            ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name());
-            ClientState clientState = makeFakeClientState();
-            CQLStatement prepared = QueryProcessor.getStatement(query, clientState).statement;
-            List<ByteBuffer> boundBBValues = new ArrayList<>();
-            for (Object boundValue : boundValues)
-            {
-                boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue));
-            }
+        return instance.sync(() -> executeInternal(query, consistencyLevelOrigin, boundValues)).call();
+    }
 
-            ResultMessage res = prepared.execute(QueryState.forInternalCalls(),
-                                                 QueryOptions.create(consistencyLevel,
-                                                                     boundBBValues,
-                                                                     false,
-                                                                     Integer.MAX_VALUE,
-                                                                     null,
-                                                                     null,
-                                                                     ProtocolVersion.CURRENT),
-                                                  System.nanoTime());
-
-            if (res != null && res.kind == ResultMessage.Kind.ROWS)
+    public Future<Object[][]> asyncExecuteWithTracing(UUID sessionId, String query, Enum<?> consistencyLevelOrigin, Object... boundValues)
+    {
+        return instance.async(() -> {
+            try
             {
-                return RowUtil.toObjects((ResultMessage.Rows) res);
+                Tracing.instance.newSession(sessionId, Collections.emptyMap());
+                return executeInternal(query, consistencyLevelOrigin, boundValues);
             }
-            else
+            finally
             {
-                return new Object[][]{};
+                Tracing.instance.stopSession();
             }
         }).call();
     }
 
+    private Object[][] executeInternal(String query, Enum<?> consistencyLevelOrigin, Object[] boundValues)
+    {
+        ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name());
+        ClientState clientState = makeFakeClientState();
+        CQLStatement prepared = QueryProcessor.getStatement(query, clientState).statement;
+        List<ByteBuffer> boundBBValues = new ArrayList<>();
+        for (Object boundValue : boundValues)
+        {
+            boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue));
+        }
+
+        ResultMessage res = prepared.execute(QueryState.forInternalCalls(),
+                                             QueryOptions.create(consistencyLevel,
+                                                                 boundBBValues,
+                                                                 false,
+                                                                 Integer.MAX_VALUE,
+                                                                 null,
+                                                                 null,
+                                                                 ProtocolVersion.CURRENT),
+                                             System.nanoTime());
+
+        if (res != null && res.kind == ResultMessage.Kind.ROWS)
+            return RowUtil.toObjects((ResultMessage.Rows) res);
+        else
+            return new Object[][]{};
+    }
+
+    public Object[][] executeWithTracing(UUID sessionId, String query, Enum<?> consistencyLevelOrigin, Object... boundValues)
+    {
+        return IsolatedExecutor.waitOn(asyncExecuteWithTracing(sessionId, query, consistencyLevelOrigin, boundValues));
+    }
+
     @Override
     public Iterator<Object[]> executeWithPaging(String query, Enum<?> consistencyLevelOrigin, int pageSize, Object... boundValues)
     {
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java b/test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java
new file mode 100644
index 0000000000..35e2903a17
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import java.net.InetAddress;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.locator.AbstractNetworkTopologySnitch;
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+public class DistributedTestSnitch extends AbstractNetworkTopologySnitch
+{
+    private static NetworkTopology mapping = null;
+
+    public String getRack(InetAddress endpoint)
+    {
+        assert mapping != null : "network topology must be assigned before using snitch";
+        int storage_port = Config.getOverrideLoadConfig().get().storage_port;
+        return mapping.localRack(InetAddressAndPort.getByAddressOverrideDefaults(endpoint, storage_port));
+    }
+
+    public String getRack(InetAddressAndPort endpoint)
+    {
+        assert mapping != null : "network topology must be assigned before using snitch";
+        return mapping.localRack(endpoint);
+    }
+
+    public String getDatacenter(InetAddress endpoint)
+    {
+        assert mapping != null : "network topology must be assigned before using snitch";
+        int storage_port = Config.getOverrideLoadConfig().get().storage_port;
+        return mapping.localDC(InetAddressAndPort.getByAddressOverrideDefaults(endpoint, storage_port));
+    }
+
+    public String getDatacenter(InetAddressAndPort endpoint)
+    {
+        assert mapping != null : "network topology must be assigned before using snitch";
+        return mapping.localDC(endpoint);
+    }
+
+    static void assign(NetworkTopology newMapping)
+    {
+        mapping = new NetworkTopology(newMapping);
+    }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
index 784eb3fe6e..19e28722da 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
@@ -21,6 +21,7 @@ package org.apache.cassandra.distributed.impl;
 import java.io.File;
 import java.io.IOException;
 import java.net.InetAddress;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -72,7 +73,6 @@ import org.apache.cassandra.io.util.DataInputBuffer;
 import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.net.IMessageSink;
-import org.apache.cassandra.net.MessageDeliveryTask;
 import org.apache.cassandra.net.MessageIn;
 import org.apache.cassandra.net.MessageOut;
 import org.apache.cassandra.net.MessagingService;
@@ -83,10 +83,13 @@ import org.apache.cassandra.service.QueryState;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.streaming.StreamCoordinator;
 import org.apache.cassandra.streaming.StreamSession;
+import org.apache.cassandra.tracing.TraceState;
+import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.transport.messages.ResultMessage;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.NanoTimeToCurrentTimeMillis;
 import org.apache.cassandra.utils.Throwables;
+import org.apache.cassandra.utils.UUIDGen;
 import org.apache.cassandra.utils.concurrent.Ref;
 import org.apache.cassandra.utils.memory.BufferPool;
 
@@ -246,6 +249,28 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
                 InetAddressAndPort toFull = lookupAddressAndPort.apply(to);
                 int version = MessagingService.instance().getVersion(to);
 
+                // Tracing logic - similar to org.apache.cassandra.net.OutboundTcpConnection.writeConnected
+                byte[] sessionBytes = (byte[]) messageOut.parameters.get(Tracing.TRACE_HEADER);
+                if (sessionBytes != null)
+                {
+                    UUID sessionId = UUIDGen.getUUID(ByteBuffer.wrap(sessionBytes));
+                    TraceState state = Tracing.instance.get(sessionId);
+                    String message = String.format("Sending %s message to %s", messageOut.verb, toFull.address);
+                    // session may have already finished; see CASSANDRA-5668
+                    if (state == null)
+                    {
+                        byte[] traceTypeBytes = (byte[]) messageOut.parameters.get(Tracing.TRACE_TYPE);
+                        Tracing.TraceType traceType = traceTypeBytes == null ? Tracing.TraceType.QUERY : Tracing.TraceType.deserialize(traceTypeBytes[0]);
+                        Tracing.instance.trace(ByteBuffer.wrap(sessionBytes), message, traceType.getTTL());
+                    }
+                    else
+                    {
+                        state.trace(message);
+                        if (messageOut.verb == MessagingService.Verb.REQUEST_RESPONSE)
+                            Tracing.instance.doneWithNonLocalSession(state);
+                    }
+                }
+
                 out.writeInt(MessagingService.PROTOCOL_MAGIC);
                 out.writeInt(id);
                 long timestamp = System.currentTimeMillis();
@@ -319,6 +344,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
             {
                 mkdirs();
 
+                assert config.networkTopology().contains(config.broadcastAddressAndPort());
+                DistributedTestSnitch.assign(config.networkTopology());
+
                 DatabaseDescriptor.daemonInitialization();
                 DatabaseDescriptor.createAllDirectories();
 
@@ -371,6 +399,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
                     initializeRing(cluster);
                 }
 
+                StorageService.instance.ensureTraceKeyspace();
+
                 SystemKeyspace.finishStartup();
 
                 if (!FBUtilities.getBroadcastAddress().equals(broadcastAddressAndPort().address))
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java
index 0ef5a69113..ff3aced3d8 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java
@@ -37,7 +37,8 @@ public class InstanceClassLoader extends URLClassLoader
                     Pair.class,
                     InetAddressAndPort.class,
                     ParameterizedClass.class,
-                    IInvokableInstance.class
+                    IInvokableInstance.class,
+                    NetworkTopology.class
             })
             .map(Class::getName)
             .collect(Collectors.toSet());
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
index a21506219f..6c3b839780 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
@@ -24,7 +24,6 @@ import org.apache.cassandra.distributed.api.Feature;
 import org.apache.cassandra.distributed.api.IInstanceConfig;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.locator.SimpleSeedProvider;
-import org.apache.cassandra.locator.SimpleSnitch;
 
 import java.io.File;
 import java.lang.reflect.Field;
@@ -43,11 +42,14 @@ public class InstanceConfig implements IInstanceConfig
     public final int num;
     public int num() { return num; }
 
+    private final NetworkTopology networkTopology;
+    public NetworkTopology networkTopology() { return networkTopology; }
+
     public final UUID hostId;
     public UUID hostId() { return hostId; }
     private final Map<String, Object> params = new TreeMap<>();
 
-    private EnumSet featureFlags;
+    private final EnumSet featureFlags;
 
     private volatile InetAddressAndPort broadcastAddressAndPort;
 
@@ -69,6 +71,7 @@ public class InstanceConfig implements IInstanceConfig
     }
 
     private InstanceConfig(int num,
+                           NetworkTopology networkTopology,
                            String broadcast_address,
                            String listen_address,
                            String broadcast_rpc_address,
@@ -81,6 +84,7 @@ public class InstanceConfig implements IInstanceConfig
                            String initial_token)
     {
         this.num = num;
+        this.networkTopology = networkTopology;
         this.hostId = java.util.UUID.randomUUID();
         this    .set("broadcast_address", broadcast_address)
                 .set("listen_address", listen_address)
@@ -101,8 +105,8 @@ public class InstanceConfig implements IInstanceConfig
                 .set("concurrent_compactors", 1)
                 .set("memtable_heap_space_in_mb", 10)
                 .set("commitlog_sync", "batch")
-                .set("storage_port", 7010)
-                .set("endpoint_snitch", SimpleSnitch.class.getName())
+                .set("storage_port", 7012)
+                .set("endpoint_snitch", DistributedTestSnitch.class.getName())
                 .set("seed_provider", new ParameterizedClass(SimpleSeedProvider.class.getName(),
                         Collections.singletonMap("seeds", "127.0.0.1")))
                 // legacy parameters
@@ -113,9 +117,11 @@ public class InstanceConfig implements IInstanceConfig
     private InstanceConfig(InstanceConfig copy)
     {
         this.num = copy.num;
+        this.networkTopology = new NetworkTopology(copy.networkTopology);
         this.params.putAll(copy.params);
         this.hostId = copy.hostId;
         this.featureFlags = copy.featureFlags;
+        this.broadcastAddressAndPort = copy.broadcastAddressAndPort;
     }
 
     public InstanceConfig with(Feature featureFlag)
@@ -192,11 +198,7 @@ public class InstanceConfig implements IInstanceConfig
         {
             valueField.set(writeToConfig, value);
         }
-        catch (IllegalAccessException e)
-        {
-            throw new IllegalStateException(e);
-        }
-        catch (IllegalArgumentException e)
+        catch (IllegalAccessException | IllegalArgumentException e)
         {
             throw new IllegalStateException(e);
         }
@@ -217,14 +219,14 @@ public class InstanceConfig implements IInstanceConfig
         return (String)params.get(name);
     }
 
-    public static InstanceConfig generate(int nodeNum, int subnet, File root, String token)
+    public static InstanceConfig generate(int nodeNum, String ipAddress, NetworkTopology networkTopology, File root, String token)
     {
-        String ipPrefix = "127.0." + subnet + ".";
         return new InstanceConfig(nodeNum,
-                                  ipPrefix + nodeNum,
-                                  ipPrefix + nodeNum,
-                                  ipPrefix + nodeNum,
-                                  ipPrefix + nodeNum,
+                                  networkTopology,
+                                  ipAddress,
+                                  ipAddress,
+                                  ipAddress,
+                                  ipAddress,
                                   String.format("%s/node%d/saved_caches", root, nodeNum),
                                   new String[] { String.format("%s/node%d/data", root, nodeNum) },
                                   String.format("%s/node%d/commitlog", root, nodeNum),
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java
index 1d26c5dec3..248155a7e2 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java
@@ -77,7 +77,7 @@ public class IsolatedExecutor implements IIsolatedExecutor
             return t;
         };
         ExecutorService shutdownExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 0, TimeUnit.SECONDS,
-                                                                  new LinkedBlockingQueue<Runnable>(), threadFactory);
+                                                                  new LinkedBlockingQueue<>(), threadFactory);
         return shutdownExecutor.submit(() -> {
             try
             {
@@ -165,7 +165,7 @@ public class IsolatedExecutor implements IIsolatedExecutor
     public static Object deserializeOneObject(byte[] bytes)
     {
         try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
-             ObjectInputStream ois = new ObjectInputStream(bais);)
+             ObjectInputStream ois = new ObjectInputStream(bais))
         {
             return ois.readObject();
         }
@@ -190,7 +190,7 @@ public class IsolatedExecutor implements IIsolatedExecutor
         }
     }
 
-    private static <T> T waitOn(Future<T> f)
+    public static <T> T waitOn(Future<T> f)
     {
         try
         {
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/NetworkTopology.java b/test/distributed/org/apache/cassandra/distributed/impl/NetworkTopology.java
new file mode 100644
index 0000000000..1176b320bc
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/impl/NetworkTopology.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.utils.Pair;
+
+public class NetworkTopology
+{
+    private final Map<InetAddressAndPort, Pair<String, String>> map;
+
+    private NetworkTopology() {
+        map = new HashMap<>();
+    }
+
+    @SuppressWarnings("WeakerAccess")
+    public NetworkTopology(NetworkTopology networkTopology)
+    {
+        map = new HashMap<>(networkTopology.map);
+    }
+
+    public static NetworkTopology build(String ipPrefix, int broadcastPort, Map<Integer, Pair<String, String>> nodeIdTopology)
+    {
+        final NetworkTopology topology = new NetworkTopology();
+
+        for (int nodeId = 1; nodeId <= nodeIdTopology.size(); nodeId++)
+        {
+            String broadcastAddress = ipPrefix + nodeId;
+
+            try
+            {
+                Pair<String,String> dcAndRack = nodeIdTopology.get(nodeId);
+                if (dcAndRack == null)
+                    throw new IllegalStateException("nodeId " + nodeId + "not found in instanceMap");
+
+                InetAddressAndPort broadcastAddressAndPort = InetAddressAndPort.getByAddressOverrideDefaults(
+                    InetAddress.getByName(broadcastAddress), broadcastPort);
+                topology.put(broadcastAddressAndPort, dcAndRack);
+            }
+            catch (UnknownHostException e)
+            {
+                throw new ConfigurationException("Unknown broadcast_address '" + broadcastAddress + '\'', false);
+            }
+        }
+        return topology;
+    }
+
+    public Pair<String, String> put(InetAddressAndPort key, Pair<String, String> value)
+    {
+        return map.put(key, value);
+    }
+
+    public String localRack(InetAddressAndPort key)
+    {
+        return map.get(key).right;
+    }
+
+    public String localDC(InetAddressAndPort key)
+    {
+        return map.get(key).left;
+    }
+
+    public boolean contains(InetAddressAndPort key)
+    {
+        return map.containsKey(key);
+    }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/TracingUtil.java b/test/distributed/org/apache/cassandra/distributed/impl/TracingUtil.java
new file mode 100644
index 0000000000..d671e553d9
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/impl/TracingUtil.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.impl;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.UUID;
+
+import org.apache.cassandra.db.ConsistencyLevel;
+
+
+/**
+ * Utilities for accessing the system_traces table from in-JVM dtests
+ */
+public class TracingUtil
+{
+    /**
+     * Represents an entry from system_traces
+     */
+    public static class TraceEntry
+    {
+        public final UUID sessionId;
+        public final UUID eventId;
+        public final String activity;
+        public final InetAddress source;
+        public final int sourceElapsed;
+        public final String thread;
+
+        private TraceEntry(UUID sessionId, UUID eventId, String activity, InetAddress sourceIP, int sourceElapsed, String thread)
+        {
+            this.sessionId = sessionId;
+            this.eventId = eventId;
+            this.activity = activity;
+            this.source = sourceIP;
+            this.sourceElapsed = sourceElapsed;
+            this.thread = thread;
+        }
+
+        static TraceEntry fromRowResultObjects(Object[] objects)
+        {
+            return new TraceEntry((UUID) objects[0],
+                                  (UUID) objects[1],
+                                  (String) objects[2],
+                                  (InetAddress) objects[3],
+                                  (Integer) objects[4],
+                                  (String) objects[5]);
+        }
+    }
+
+    public static List<TraceEntry> getTrace(AbstractCluster cluster, UUID sessionId)
+    {
+        return getTrace(cluster, sessionId, ConsistencyLevel.ALL);
+    }
+
+    public static List<TraceEntry> getTrace(AbstractCluster cluster, UUID sessionId, ConsistencyLevel cl)
+    {
+        Object[][] result = cluster.coordinator(1).execute(
+        "SELECT session_id, event_id, activity, source, source_elapsed, thread " +
+        "FROM system_traces.events WHERE session_id = ?", cl, sessionId);
+
+        List<TraceEntry> traces = new LinkedList<>();
+        for (Object[] r : result)
+        {
+            traces.add(TraceEntry.fromRowResultObjects(r));
+        }
+        return traces;
+    }
+
+    public static List<TraceEntry> getTraces(AbstractCluster cluster)
+    {
+        return getTraces(cluster, ConsistencyLevel.ALL);
+    }
+
+    public static List<TraceEntry> getTraces(AbstractCluster cluster, ConsistencyLevel cl)
+    {
+        Object[][] result = cluster.coordinator(1).execute(
+        "SELECT session_id, event_id, activity, source, source_elapsed, thread " +
+        "FROM system_traces.events", cl);
+
+        List<TraceEntry> traces = new ArrayList<>();
+        for (Object[] r : result)
+        {
+            traces.add(TraceEntry.fromRowResultObjects(r));
+        }
+        return traces;
+    }
+
+    // Set up the wait for tracing time system property, returning the previous value.
+    // Handles being called again to reset with the original value, replacing the null
+    // with the default value.
+    public static String setWaitForTracingEventTimeoutSecs(String timeoutInSeconds)
+    {
+        return System.setProperty("cassandra.wait_for_tracing_events_timeout_secs",
+                                  timeoutInSeconds == null ? "0" : timeoutInSeconds);
+
+    }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java b/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java
index 238e9e7efa..ac5028611c 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java
@@ -85,7 +85,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase
             assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"));
 
             assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
-                                                     ConsistencyLevel.QUORUM),
+                                                     ConsistencyLevel.ALL), // ensure node3 in preflist
                        row(1, 1, 1));
 
             // Verify that data got repaired to the third node
diff --git a/test/distributed/org/apache/cassandra/distributed/test/MessageForwardingTest.java b/test/distributed/org/apache/cassandra/distributed/test/MessageForwardingTest.java
new file mode 100644
index 0000000000..72928e4166
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/MessageForwardingTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.Future;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.impl.IsolatedExecutor;
+import org.apache.cassandra.distributed.impl.TracingUtil;
+import org.apache.cassandra.utils.UUIDGen;
+
+public class MessageForwardingTest extends DistributedTestBase
+{
+    @Test
+    public void mutationsForwardedToAllReplicasTest()
+    {
+        String originalTraceTimeout = TracingUtil.setWaitForTracingEventTimeoutSecs("1");
+        final int numInserts = 100;
+        Map<InetAddress,Integer> commitCounts = new HashMap<>();
+
+        try (Cluster cluster = init(Cluster.build()
+                                           .withDC("dc0", 1)
+                                           .withDC("dc1", 3)
+                                           .start()))
+        {
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v text, PRIMARY KEY (pk, ck))");
+
+            cluster.forEach(instance -> commitCounts.put(instance.broadcastAddressAndPort().address, 0));
+            final UUID sessionId = UUIDGen.getTimeUUID();
+            Stream<Future<Object[][]>> inserts = IntStream.range(0, numInserts).mapToObj((idx) ->
+                cluster.coordinator(1).asyncExecuteWithTracing(sessionId,
+                                                         "INSERT INTO " + KEYSPACE + ".tbl(pk,ck,v) VALUES (1, 1, 'x')",
+                                                         ConsistencyLevel.ALL)
+            );
+
+            // Wait for each of the futures to complete before checking the traces, don't care
+            // about the result so
+            //noinspection ResultOfMethodCallIgnored
+            inserts.map(IsolatedExecutor::waitOn).count();
+
+            cluster.forEach(instance -> commitCounts.put(instance.broadcastAddressAndPort().address, 0));
+            List<TracingUtil.TraceEntry> traces = TracingUtil.getTrace(cluster, sessionId, ConsistencyLevel.ALL);
+            traces.forEach(traceEntry -> {
+                if (traceEntry.activity.contains("Appending to commitlog"))
+                {
+                    commitCounts.compute(traceEntry.source, (k, v) -> (v != null ? v : 0) + 1);
+                }
+            });
+
+            // Check that each node received the forwarded messages once (and only once)
+            commitCounts.forEach((source, count) ->
+                                 Assert.assertEquals(source + " appending to commitlog traces",
+                                                     (long) numInserts, (long) count));
+        }
+        catch (IOException e)
+        {
+            Assert.fail("Threw exception: " + e);
+        }
+        finally
+        {
+            TracingUtil.setWaitForTracingEventTimeoutSecs(originalTraceTimeout);
+        }
+    }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java b/test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java
new file mode 100644
index 0000000000..2c4f9c8904
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import java.util.Collections;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.utils.Pair;
+
+public class NetworkTopologyTest extends DistributedTestBase
+{
+    @Test
+    public void namedDcTest() throws Throwable
+    {
+        try (Cluster cluster = Cluster.build()
+                                      .withNodeIdTopology(Collections.singletonMap(1, Pair.create("somewhere", "rack0")))
+                                      .withRack("elsewhere", "firstrack", 1)
+                                      .withRack("elsewhere", "secondrack", 2)
+                                      .withDC("nearthere", 4)
+                                      .start())
+        {
+            Assert.assertEquals(1, cluster.stream("somewhere").count());
+            Assert.assertEquals(1, cluster.stream("elsewhere", "firstrack").count());
+            Assert.assertEquals(2, cluster.stream("elsewhere", "secondrack").count());
+            Assert.assertEquals(3, cluster.stream("elsewhere").count());
+            Assert.assertEquals(4, cluster.stream("nearthere").count());
+
+            Set<IInstance> expect = cluster.stream().collect(Collectors.toSet());
+            Set<IInstance> result = Stream.concat(Stream.concat(cluster.stream("somewhere"),
+                                                                cluster.stream("elsewhere")),
+                                                  cluster.stream("nearthere")).collect(Collectors.toSet());
+            Assert.assertEquals(expect, result);
+        }
+    }
+
+    @Test
+    public void automaticNamedDcTest() throws Throwable
+
+    {
+        try (Cluster cluster = Cluster.build()
+                                      .withRacks(2, 1, 3)
+                                      .start())
+        {
+            Assert.assertEquals(6, cluster.stream().count());
+            Assert.assertEquals(3, cluster.stream("datacenter1").count());
+            Assert.assertEquals(3, cluster.stream("datacenter2", "rack1").count());
+        }
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void noCountsAfterNamingDCsTest()
+    {
+        Cluster.build()
+               .withDC("nameddc", 1)
+               .withDCs(1);
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void mustProvideNodeCountBeforeWithDCsTest()
+    {
+        Cluster.build()
+               .withDCs(1);
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void noEmptyNodeIdTopologyTest()
+    {
+        Cluster.build().withNodeIdTopology(Collections.emptyMap());
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void noHolesInNodeIdTopologyTest()
+    {
+        Cluster.build().withNodeIdTopology(Collections.singletonMap(2, Pair.create("doomed", "rack")));
+    }
+}