diff --git a/.build/parent-pom-template.xml b/.build/parent-pom-template.xml
index 7200bb4..e4b63bf 100644
--- a/.build/parent-pom-template.xml
+++ b/.build/parent-pom-template.xml
@@ -489,7 +489,7 @@
       <dependency>
         <groupId>org.apache.cassandra</groupId>
         <artifactId>dtest-api</artifactId>
-        <version>0.0.13</version>
+        <version>0.0.15</version>
         <scope>test</scope>
       </dependency>
       <dependency>
diff --git a/build.xml b/build.xml
index a3ef1fc..3e0a8e9 100644
--- a/build.xml
+++ b/build.xml
@@ -207,6 +207,7 @@
         <string>--add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED</string>
         <string>--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED</string>
         <string>--add-exports java.rmi/sun.rmi.server=ALL-UNNAMED</string>
+        <string>--add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED</string>
         <string>--add-exports java.sql/java.sql=ALL-UNNAMED</string>
 
         <string>--add-opens java.base/java.lang.module=ALL-UNNAMED</string>
@@ -237,6 +238,7 @@
         <string>--add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED</string>
         <string>--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED</string>
         <string>--add-exports java.rmi/sun.rmi.server=ALL-UNNAMED</string>
+        <string>--add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED</string>
         <string>--add-exports java.sql/java.sql=ALL-UNNAMED</string>
         <string>--add-exports java.base/java.lang.ref=ALL-UNNAMED</string>
         <string>--add-exports jdk.unsupported/sun.misc=ALL-UNNAMED</string>
@@ -263,6 +265,8 @@
         <string>--add-opens java.base/java.math=ALL-UNNAMED</string>
         <string>--add-opens java.base/java.lang.reflect=ALL-UNNAMED</string>
         <string>--add-opens java.base/java.net=ALL-UNNAMED</string>
+
+        <string>--add-opens java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED</string>
     </resources>
     <pathconvert property="_jvm_args_concat2" refid="_jvm17_arg_items" pathsep=" "/>
     <condition property="java17-jvmargs" value="${_jvm_args_concat2}" else="">
@@ -307,7 +311,8 @@
     </condition>
 
     <!-- needed to compile org.apache.cassandra.utils.JMXServerUtils -->
-    <condition property="jdk11plus-javac-exports" value="--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED" else="">
+    <!-- needed to compile org.apache.cassandra.distributed.impl.Instance-->
+    <condition property="jdk11plus-javac-exports" value="--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED" else="">
         <not>
             <equals arg1="${ant.java.version}" arg2="1.8"/>
         </not>
@@ -1035,6 +1040,7 @@
      <src path="${test.simulator-asm.src}"/>
      <src path="${test.simulator-bootstrap.src}"/>
      <src path="${test.simulator-test.src}"/>
+     <compilerarg line="${jdk11plus-javac-exports}"/>
     </javac>
 
     <!-- Non-java resources needed by the test suite -->
@@ -1811,6 +1817,24 @@
       <echo>"IDE configuration updated for use with JDK11"</echo>
   </target>
 
+  <target name="_maybe_update_idea_to_java17" depends="init" if="java.version.17">
+        <replace file="${eclipse.project.name}.iml" token="JDK_1_8" value="JDK_17"/>
+        <replace file=".idea/misc.xml" token="JDK_1_8" value="JDK_17"/>
+        <replace file=".idea/misc.xml" token="1.8" value="17"/>
+        <replaceregexp file=".idea/workspace.xml"
+                       match="name=&quot;VM_PARAMETERS&quot; value=&quot;(.*)"
+                       replace="name=&quot;VM_PARAMETERS&quot; value=&quot;\1 ${java17-jvmargs}"
+                       byline="true"/>
+
+        <echo file=".idea/compiler.xml"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="JavacSettings">
+    <option name="ADDITIONAL_OPTIONS_STRING" value="--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED" />
+  </component>
+</project>]]></echo>
+        <echo>"IDE configuration updated for use with JDK17"</echo>
+    </target>
+
   <!-- Generate IDEA project description files -->
   <target name="generate-idea-files" depends="init,resolver-dist-lib,gen-cql3-grammar,generate-jflex-java,createVersionPropFile" description="Generate IDEA files">
     <mkdir dir=".idea"/>
@@ -1831,6 +1855,7 @@
   </component>
 </project>]]></echo>
       <antcall target="_maybe_update_idea_to_java11"/>
+      <antcall target="_maybe_update_idea_to_java17"/>
   </target>
 
   <!-- Generate Eclipse project description files -->
diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
index 78e1992..4daa825 100644
--- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
+++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
@@ -200,6 +200,8 @@
     DRAIN_EXECUTOR_TIMEOUT_MS("cassandra.drain_executor_timeout_ms", convertToString(TimeUnit.MINUTES.toMillis(5))),
     DROP_OVERSIZED_READ_REPAIR_MUTATIONS("cassandra.drop_oversized_readrepair_mutations"),
     DTEST_API_LOG_TOPOLOGY("cassandra.dtest.api.log.topology"),
+    /** This property indicates if the code is running under the in-jvm dtest framework */
+    DTEST_IS_IN_JVM_DTEST("org.apache.cassandra.dtest.is_in_jvm_dtest"),
     ENABLE_DC_LOCAL_COMMIT("cassandra.enable_dc_local_commit", "true"),
     /**
      * Whether {@link org.apache.cassandra.db.ConsistencyLevel#NODE_LOCAL} should be allowed.
@@ -263,6 +265,11 @@
      */
     JAVA_LIBRARY_PATH("java.library.path"),
     /**
+     * Controls the distributed garbage collector lease time for JMX objects.
+     * Should only be set by in-jvm dtests.
+     */
+    JAVA_RMI_DGC_LEASE_VALUE_IN_JVM_DTEST("java.rmi.dgc.leaseValue"),
+    /**
      * The value of this property represents the host name string
      * that should be associated with remote stubs for locally created remote objects,
      * in order to allow clients to invoke methods on the remote object.
@@ -421,6 +428,11 @@
     /** Platform word size sun.arch.data.model. Examples: "32", "64", "unknown"*/
     SUN_ARCH_DATA_MODEL("sun.arch.data.model"),
     SUN_JAVA_COMMAND("sun.java.command", ""),
+    /**
+     * Controls the JMX server threadpool keap-alive time.
+     * Should only be set by in-jvm dtests.
+     */
+    SUN_RMI_TRANSPORT_TCP_THREADKEEPALIVETIME("sun.rmi.transport.tcp.threadKeepAliveTime"),
     SUN_STDERR_ENCODING("sun.stderr.encoding"),
     SUN_STDOUT_ENCODING("sun.stdout.encoding"),
     SUPERUSER_SETUP_DELAY_MS("cassandra.superuser_setup_delay_ms", "10000"),
diff --git a/src/java/org/apache/cassandra/service/GCInspector.java b/src/java/org/apache/cassandra/service/GCInspector.java
index c290d9e..8f92215 100644
--- a/src/java/org/apache/cassandra/service/GCInspector.java
+++ b/src/java/org/apache/cassandra/service/GCInspector.java
@@ -148,18 +148,16 @@
 
     public GCInspector()
     {
-        MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
-
         try
         {
             ObjectName gcName = new ObjectName(ManagementFactory.GARBAGE_COLLECTOR_MXBEAN_DOMAIN_TYPE + ",*");
-            for (ObjectName name : mbs.queryNames(gcName, null))
+            for (ObjectName name : MBeanWrapper.instance.queryNames(gcName, null))
             {
-                GarbageCollectorMXBean gc = ManagementFactory.newPlatformMXBeanProxy(mbs, name.getCanonicalName(), GarbageCollectorMXBean.class);
+                GarbageCollectorMXBean gc = ManagementFactory.newPlatformMXBeanProxy(MBeanWrapper.instance.getMBeanServer(), name.getCanonicalName(), GarbageCollectorMXBean.class);
                 gcStates.put(gc.getName(), new GCState(gc, assumeGCIsPartiallyConcurrent(gc), assumeGCIsOldGen(gc)));
             }
             ObjectName me = new ObjectName(MBEAN_NAME);
-            if (!mbs.isRegistered(me))
+            if (!MBeanWrapper.instance.isRegistered(me))
                 MBeanWrapper.instance.registerMBean(this, new ObjectName(MBEAN_NAME));
         }
         catch (MalformedObjectNameException | IOException e)
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 1d4a0b1..5e66718 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -987,7 +987,8 @@
         completeInitialization();
     }
 
-    private void completeInitialization()
+    @VisibleForTesting
+    public void completeInitialization()
     {
         if (!initialized)
             registerMBeans();
diff --git a/src/java/org/apache/cassandra/utils/JMXServerUtils.java b/src/java/org/apache/cassandra/utils/JMXServerUtils.java
index ba9c704..45caf6e 100644
--- a/src/java/org/apache/cassandra/utils/JMXServerUtils.java
+++ b/src/java/org/apache/cassandra/utils/JMXServerUtils.java
@@ -29,12 +29,14 @@
 import java.net.InetAddress;
 import java.rmi.AccessException;
 import java.rmi.AlreadyBoundException;
+import java.rmi.NoSuchObjectException;
 import java.rmi.NotBoundException;
 import java.rmi.Remote;
 import java.rmi.RemoteException;
 import java.rmi.registry.Registry;
 import java.rmi.server.RMIClientSocketFactory;
 import java.rmi.server.RMIServerSocketFactory;
+import java.rmi.server.UnicastRemoteObject;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
@@ -257,7 +259,8 @@
         return env;
     }
 
-    private static void logJmxServiceUrl(InetAddress serverAddress, int port)
+    @VisibleForTesting
+    public static void logJmxServiceUrl(InetAddress serverAddress, int port)
     {
         String urlTemplate = "service:jmx:rmi://%1$s/jndi/rmi://%1$s:%2$d/jmxrmi";
         String hostName;
@@ -330,11 +333,11 @@
      * Better to use the internal API than re-invent the wheel.
      */
     @SuppressWarnings("restriction")
-    private static class JmxRegistry extends sun.rmi.registry.RegistryImpl {
+    public static class JmxRegistry extends sun.rmi.registry.RegistryImpl {
         private final String lookupName;
         private Remote remoteServerStub;
 
-        JmxRegistry(final int port,
+        public JmxRegistry(final int port,
                     final RMIClientSocketFactory csf,
                     RMIServerSocketFactory ssf,
                     final String lookupName) throws RemoteException {
@@ -367,5 +370,24 @@
         public void setRemoteServerStub(Remote remoteServerStub) {
             this.remoteServerStub = remoteServerStub;
         }
+
+        /**
+         * Closes the underlying JMX registry by unexporting this instance.
+         * There is no reason to do this except for in-jvm dtests where we need
+         * to stop the registry, so we can start with a clean slate for future cluster
+         * builds, and the superclass never expects to be shut down and therefore doesn't
+         * handle this edge case at all.
+         */
+        @VisibleForTesting
+        public void close() {
+            try
+            {
+                UnicastRemoteObject.unexportObject(this, true);
+            }
+            catch (NoSuchObjectException ignored)
+            {
+                // Ignore if it's already unexported
+            }
+        }
     }
 }
diff --git a/src/java/org/apache/cassandra/utils/MBeanWrapper.java b/src/java/org/apache/cassandra/utils/MBeanWrapper.java
index f003e1a..0ef342d 100644
--- a/src/java/org/apache/cassandra/utils/MBeanWrapper.java
+++ b/src/java/org/apache/cassandra/utils/MBeanWrapper.java
@@ -19,15 +19,21 @@
 package org.apache.cassandra.utils;
 
 import java.lang.management.ManagementFactory;
+import java.util.Collections;
+import java.util.Set;
+import java.util.UUID;
 import java.util.function.Consumer;
 import javax.management.MBeanServer;
+import javax.management.MBeanServerFactory;
 import javax.management.MalformedObjectNameException;
 import javax.management.ObjectName;
+import javax.management.QueryExp;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION;
+import static org.apache.cassandra.config.CassandraRelevantProperties.DTEST_IS_IN_JVM_DTEST;
 import static org.apache.cassandra.config.CassandraRelevantProperties.MBEAN_REGISTRATION_CLASS;
 
 /**
@@ -36,23 +42,46 @@
  */
 public interface MBeanWrapper
 {
-    static final Logger logger = LoggerFactory.getLogger(MBeanWrapper.class);
+    Logger logger = LoggerFactory.getLogger(MBeanWrapper.class);
 
-    static final MBeanWrapper instance = create();
+    MBeanWrapper instance = create();
 
     static MBeanWrapper create()
     {
+        // If we're running in the in-jvm dtest environment, always use the delegating
+        // mbean wrapper even if we start off with no-op, so it can be switched later
+        if (DTEST_IS_IN_JVM_DTEST.getBoolean())
+        {
+            return new DelegatingMbeanWrapper(getMBeanWrapper());
+        }
+
+        return getMBeanWrapper();
+    }
+
+    static MBeanWrapper getMBeanWrapper()
+    {
         if (ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.getBoolean())
+        {
             return new NoOpMBeanWrapper();
+        }
 
         String klass = MBEAN_REGISTRATION_CLASS.getString();
         if (klass == null)
-            return new PlatformMBeanWrapper();
+        {
+            if (DTEST_IS_IN_JVM_DTEST.getBoolean())
+            {
+                return new NoOpMBeanWrapper();
+            }
+            else
+            {
+                return new PlatformMBeanWrapper();
+            }
+        }
         return FBUtilities.construct(klass, "mbean");
     }
 
     // Passing true for graceful will log exceptions instead of rethrowing them
-    public void registerMBean(Object obj, ObjectName mbeanName, OnException onException);
+    void registerMBean(Object obj, ObjectName mbeanName, OnException onException);
     default void registerMBean(Object obj, ObjectName mbeanName)
     {
         registerMBean(obj, mbeanName, OnException.THROW);
@@ -62,7 +91,9 @@
     {
         ObjectName name = create(mbeanName, onException);
         if (name == null)
+        {
             return;
+        }
         registerMBean(obj, name, onException);
     }
     default void registerMBean(Object obj, String mbeanName)
@@ -70,7 +101,7 @@
         registerMBean(obj, mbeanName, OnException.THROW);
     }
 
-    public boolean isRegistered(ObjectName mbeanName, OnException onException);
+    boolean isRegistered(ObjectName mbeanName, OnException onException);
     default boolean isRegistered(ObjectName mbeanName)
     {
         return isRegistered(mbeanName, OnException.THROW);
@@ -80,7 +111,9 @@
     {
         ObjectName name = create(mbeanName, onException);
         if (name == null)
+        {
             return false;
+        }
         return isRegistered(name, onException);
     }
     default boolean isRegistered(String mbeanName)
@@ -88,7 +121,7 @@
         return isRegistered(mbeanName, OnException.THROW);
     }
 
-    public void unregisterMBean(ObjectName mbeanName, OnException onException);
+    void unregisterMBean(ObjectName mbeanName, OnException onException);
     default void unregisterMBean(ObjectName mbeanName)
     {
         unregisterMBean(mbeanName, OnException.THROW);
@@ -98,7 +131,9 @@
     {
         ObjectName name = create(mbeanName, onException);
         if (name == null)
+        {
             return;
+        }
         unregisterMBean(name, onException);
     }
     default void unregisterMBean(String mbeanName)
@@ -119,7 +154,11 @@
         }
     }
 
-    static class NoOpMBeanWrapper implements MBeanWrapper
+    Set<ObjectName> queryNames(ObjectName name, QueryExp query);
+
+    MBeanServer getMBeanServer();
+
+    class NoOpMBeanWrapper implements MBeanWrapper
     {
         public void registerMBean(Object obj, ObjectName mbeanName, OnException onException) {}
         public void registerMBean(Object obj, String mbeanName, OnException onException) {}
@@ -127,9 +166,11 @@
         public boolean isRegistered(String mbeanName, OnException onException) { return false; }
         public void unregisterMBean(ObjectName mbeanName, OnException onException) {}
         public void unregisterMBean(String mbeanName, OnException onException) {}
+        public Set<ObjectName> queryNames(ObjectName name, QueryExp query) {return Collections.emptySet(); }
+        public MBeanServer getMBeanServer() { return null; }
     }
 
-    static class PlatformMBeanWrapper implements MBeanWrapper
+    class PlatformMBeanWrapper implements MBeanWrapper
     {
         private final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
         public void registerMBean(Object obj, ObjectName mbeanName, OnException onException)
@@ -168,9 +209,158 @@
                 onException.handler.accept(e);
             }
         }
+
+        public Set<ObjectName> queryNames(ObjectName name, QueryExp query)
+        {
+            return mbs.queryNames(name, query);
+        }
+
+        public MBeanServer getMBeanServer()
+        {
+            return mbs;
+        }
     }
 
-    public enum OnException
+    class InstanceMBeanWrapper implements MBeanWrapper
+    {
+        private MBeanServer mbs;
+        public final UUID id = UUID.randomUUID();
+
+        public InstanceMBeanWrapper(String hostname)
+        {
+            mbs = MBeanServerFactory.createMBeanServer(hostname + "-" + id);
+        }
+
+        public void registerMBean(Object obj, ObjectName mbeanName, OnException onException)
+        {
+            try
+            {
+                mbs.registerMBean(obj, mbeanName);
+            }
+            catch (Exception e)
+            {
+                onException.handler.accept(e);
+            }
+        }
+
+        public boolean isRegistered(ObjectName mbeanName, OnException onException)
+        {
+            try
+            {
+                return mbs.isRegistered(mbeanName);
+            }
+            catch (Exception e)
+            {
+                onException.handler.accept(e);
+            }
+            return false;
+        }
+
+        public void unregisterMBean(ObjectName mbeanName, OnException onException)
+        {
+            try
+            {
+                mbs.unregisterMBean(mbeanName);
+            }
+            catch (Exception e)
+            {
+                onException.handler.accept(e);
+            }
+        }
+
+        public Set<ObjectName> queryNames(ObjectName name, QueryExp query)
+        {
+            return mbs.queryNames(name, query);
+        }
+
+        public MBeanServer getMBeanServer()
+        {
+            return mbs;
+        }
+
+        public void close() {
+            mbs.queryNames(null, null).forEach(name -> {
+                try {
+                    if (!name.getCanonicalName().contains("MBeanServerDelegate"))
+                    {
+                        mbs.unregisterMBean(name);
+                    }
+                } catch (Throwable e) {
+                    logger.debug("Could not unregister mbean {}", name.getCanonicalName());
+                }
+            });
+            MBeanServerFactory.releaseMBeanServer(mbs);
+            mbs = null;
+        }
+    }
+
+    class DelegatingMbeanWrapper implements MBeanWrapper
+    {
+        MBeanWrapper delegate;
+
+        public DelegatingMbeanWrapper(MBeanWrapper mBeanWrapper)
+        {
+            delegate = mBeanWrapper;
+        }
+
+        public void setDelegate(MBeanWrapper wrapper) {
+            delegate = wrapper;
+        }
+
+        public MBeanWrapper getDelegate()
+        {
+            return delegate;
+        }
+        
+        public void registerMBean(Object obj, ObjectName mbeanName, OnException onException)
+        {
+            try
+            {
+                delegate.registerMBean(obj, mbeanName);
+            }
+            catch (Exception e)
+            {
+                onException.handler.accept(e);
+            }
+        }
+
+        public boolean isRegistered(ObjectName mbeanName, OnException onException)
+        {
+            try
+            {
+                return delegate.isRegistered(mbeanName);
+            }
+            catch (Exception e)
+            {
+                onException.handler.accept(e);
+            }
+            return false;
+        }
+
+        public void unregisterMBean(ObjectName mbeanName, OnException onException)
+        {
+            try
+            {
+                delegate.unregisterMBean(mbeanName);
+            }
+            catch (Exception e)
+            {
+                onException.handler.accept(e);
+            }
+        }
+
+        public Set<ObjectName> queryNames(ObjectName name, QueryExp query)
+        {
+            return delegate.queryNames(name, query);
+        }
+
+        public MBeanServer getMBeanServer()
+        {
+            return delegate.getMBeanServer();
+        }
+    }
+
+    enum OnException
     {
         THROW(e -> { throw new RuntimeException(e); }),
         LOG(e -> { logger.error("Error in MBean wrapper: ", e); }),
diff --git a/src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java b/src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java
new file mode 100644
index 0000000..62ab88f
--- /dev/null
+++ b/src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.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.utils;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.net.InetAddress;
+import java.net.Socket;
+import java.rmi.server.RMIClientSocketFactory;
+import java.util.Objects;
+
+/**
+ * This class is used to override the local address the JMX client calculates when trying to connect,
+ * which can otherwise be influenced by the system property "java.rmi.server.hostname" in strange and
+ * unpredictable ways.
+ */
+public class RMIClientSocketFactoryImpl implements RMIClientSocketFactory, Serializable
+{
+    private final InetAddress localAddress;
+
+    public RMIClientSocketFactoryImpl(InetAddress localAddress)
+    {
+        this.localAddress = localAddress;
+    }
+
+    @Override
+    public Socket createSocket(String host, int port) throws IOException
+    {
+        return new Socket(localAddress, port);
+    }
+
+    @Override
+    public boolean equals(Object o)
+    {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        RMIClientSocketFactoryImpl that = (RMIClientSocketFactoryImpl) o;
+        return Objects.equals(localAddress, that.localAddress);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Objects.hash(localAddress);
+    }
+}
diff --git a/src/java/org/apache/cassandra/utils/ReflectionUtils.java b/src/java/org/apache/cassandra/utils/ReflectionUtils.java
index 801256d..bf6ad2c 100644
--- a/src/java/org/apache/cassandra/utils/ReflectionUtils.java
+++ b/src/java/org/apache/cassandra/utils/ReflectionUtils.java
@@ -30,10 +30,15 @@
 
     public static Field getModifiersField() throws NoSuchFieldException
     {
+        return getField(Field.class, "modifiers");
+    }
+
+    public static Field getField(Class<?> clazz, String fieldName) throws NoSuchFieldException
+    {
         // below code works before Java 12
         try
         {
-            return Field.class.getDeclaredField("modifiers");
+            return clazz.getDeclaredField(fieldName);
         }
         catch (NoSuchFieldException e)
         {
@@ -42,10 +47,10 @@
             {
                 Method getDeclaredFields0 = Class.class.getDeclaredMethod("getDeclaredFields0", boolean.class);
                 getDeclaredFields0.setAccessible(true);
-                Field[] fields = (Field[]) getDeclaredFields0.invoke(Field.class, false);
+                Field[] fields = (Field[]) getDeclaredFields0.invoke(clazz, false);
                 for (Field field : fields)
                 {
-                    if ("modifiers".equals(field.getName()))
+                    if (fieldName.equals(field.getName()))
                     {
                         return field;
                     }
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
index 78b14d2..79120a2 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
@@ -186,6 +186,8 @@
         private ShutdownExecutor shutdownExecutor = DEFAULT_SHUTDOWN_EXECUTOR;
 
         {
+            // Indicate that we are running in the in-jvm dtest environment
+            CassandraRelevantProperties.DTEST_IS_IN_JVM_DTEST.setBoolean(true);
             // those properties may be set for unit-test optimizations; those should not be used when running dtests
             CassandraRelevantProperties.TEST_FLUSH_LOCAL_SCHEMA_CHANGES.reset();
             CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN.reset();
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.java b/test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.java
new file mode 100644
index 0000000..0fc7425
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.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.impl;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketException;
+import java.rmi.server.RMIServerSocketFactory;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+import javax.net.ServerSocketFactory;
+
+
+/**
+ * This class is used to keep track of RMI servers created during a cluster creation so we can
+ * later close the sockets, which would otherwise be left with a thread running waiting for
+ * connections that would never show up as the server was otherwise closed.
+ */
+class CollectingRMIServerSocketFactoryImpl implements RMIServerSocketFactory
+{
+    private final InetAddress bindAddress;
+    List<ServerSocket> sockets = new ArrayList<>();
+
+    public CollectingRMIServerSocketFactoryImpl(InetAddress bindAddress)
+    {
+        this.bindAddress = bindAddress;
+    }
+
+    @Override
+    public ServerSocket createServerSocket(int pPort) throws IOException
+    {
+        ServerSocket result = ServerSocketFactory.getDefault().createServerSocket(pPort, 0, bindAddress);
+        try
+        {
+            result.setReuseAddress(true);
+        }
+        catch (SocketException e)
+        {
+            result.close();
+            throw e;
+        }
+        sockets.add(result);
+        return result;
+    }
+
+
+    public void close() throws IOException
+    {
+        for (ServerSocket socket : sockets)
+        {
+            socket.close();
+        }
+    }
+
+    @Override
+    public boolean equals(Object o)
+    {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        CollectingRMIServerSocketFactoryImpl that = (CollectingRMIServerSocketFactoryImpl) o;
+        return Objects.equals(bindAddress, that.bindAddress);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Objects.hash(bindAddress);
+    }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java b/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java
index 1ec844e..99ef272 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java
@@ -56,6 +56,11 @@
                     {
                         return 9041 + nodeNum;
                     }
+
+                    public int jmxPort(int nodeNum)
+                    {
+                        return 7199 + nodeNum;
+                    }
                 };
             }
         },
@@ -89,6 +94,11 @@
                     {
                         return 9042;
                     }
+
+                    public int jmxPort(int nodeNum)
+                    {
+                        return 7199;
+                    }
                 };
             }
         };
@@ -100,4 +110,5 @@
     abstract String ipAddress(int nodeNum);
     abstract int storagePort(int nodeNum);
     abstract int nativeTransportPort(int nodeNum);
+    abstract int jmxPort(int nodeNum);
 }
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
index 435de65..707da49 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
@@ -160,6 +160,7 @@
 import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_CASSANDRA_TESTTAG;
 import static org.apache.cassandra.distributed.api.Feature.BLANK_GOSSIP;
 import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.JMX;
 import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
 import static org.apache.cassandra.distributed.api.Feature.NETWORK;
 import static org.apache.cassandra.distributed.impl.DistributedTestSnitch.fromCassandraInetAddressAndPort;
@@ -177,6 +178,7 @@
     private volatile boolean initialized = false;
     private volatile boolean internodeMessagingStarted = false;
     private final AtomicLong startedAt = new AtomicLong();
+    private IsolatedJmx isolatedJmx;
 
     @Deprecated
     Instance(IInstanceConfig config, ClassLoader classLoader)
@@ -603,6 +605,9 @@
                                                                                                     config.networkTopology(), config.broadcastAddress());
                 DistributedTestSnitch.assign(config.networkTopology());
 
+                if (config.has(JMX))
+                    startJmx();
+
                 DatabaseDescriptor.daemonInitialization();
                 LoggingSupportFactory.getLoggingSupport().onStartup();
 
@@ -723,6 +728,7 @@
                     StorageService.instance.setNormalModeUnsafe();
                     Gossiper.instance.register(StorageService.instance);
                     StorageService.instance.startSnapshotManager();
+                    StorageService.instance.completeInitialization();
                 }
 
                 // Populate tokenMetadata for the second time,
@@ -759,6 +765,20 @@
         initialized = true;
     }
 
+    private void startJmx()
+    {
+        this.isolatedJmx = new IsolatedJmx(this, inInstancelogger);
+        isolatedJmx.startJmx();
+    }
+
+    private void stopJmx() throws IllegalAccessException, NoSuchFieldException, InterruptedException
+    {
+        if (config.has(JMX))
+        {
+            isolatedJmx.stopJmx();
+        }
+    }
+
     // Update the messaging versions for all instances
     // that have initialized their configurations.
     private static void propagateMessagingVersions(ICluster cluster)
@@ -900,6 +920,8 @@
 
             // ScheduledExecutors shuts down after MessagingService, as MessagingService may issue tasks to it.
             error = parallelRun(error, executor, () -> ScheduledExecutors.shutdownNowAndWait(1L, MINUTES));
+            
+            error = parallelRun(error, executor, this::stopJmx);
 
             // Make sure any shutdown hooks registered for DeleteOnExit are released to prevent
             // references to the instance class loaders from being held
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
index 92c56d6..3c515d5 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
@@ -41,6 +41,8 @@
 public class InstanceConfig implements IInstanceConfig
 {
     public final int num;
+    private final int jmxPort;
+
     public int num() { return num; }
 
     private final NetworkTopology networkTopology;
@@ -71,7 +73,8 @@
                            String cdc_raw_directory,
                            Collection<String> initial_token,
                            int storage_port,
-                           int native_transport_port)
+                           int native_transport_port,
+                           int jmx_port)
     {
         this.num = num;
         this.networkTopology = networkTopology;
@@ -113,6 +116,7 @@
                 // legacy parameters
                 .forceSet("commitlog_sync_batch_window_in_ms", "1");
         this.featureFlags = EnumSet.noneOf(Feature.class);
+        this.jmxPort = jmx_port;
     }
 
     private InstanceConfig(InstanceConfig copy)
@@ -124,6 +128,7 @@
         this.hostId = copy.hostId;
         this.featureFlags = copy.featureFlags;
         this.broadcastAddressAndPort = copy.broadcastAddressAndPort;
+        this.jmxPort = copy.jmxPort;
     }
 
     @Override
@@ -168,6 +173,12 @@
         return networkTopology().localDC(broadcastAddress());
     }
 
+    @Override
+    public int jmxPort()
+    {
+        return this.jmxPort;
+    }
+
     public InstanceConfig with(Feature featureFlag)
     {
         featureFlags.add(featureFlag);
@@ -267,7 +278,8 @@
                                   String.format("%s/node%d/cdc", root, nodeNum),
                                   tokens,
                                   provisionStrategy.storagePort(nodeNum),
-                                  provisionStrategy.nativeTransportPort(nodeNum));
+                                  provisionStrategy.nativeTransportPort(nodeNum),
+                                  provisionStrategy.jmxPort(nodeNum));
     }
 
     private static String[] datadirs(int datadirCount, Path root, int nodeNum)
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java
new file mode 100644
index 0000000..e19e29f
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java
@@ -0,0 +1,230 @@
+/*
+ * 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.lang.reflect.Field;
+import java.net.InetAddress;
+import java.net.MalformedURLException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+import javax.management.remote.JMXConnector;
+import javax.management.remote.JMXConnectorFactory;
+import javax.management.remote.JMXConnectorServer;
+import javax.management.remote.JMXServiceURL;
+import javax.management.remote.rmi.RMIConnectorServer;
+import javax.management.remote.rmi.RMIJRMPServerImpl;
+
+import org.slf4j.Logger;
+
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.utils.JMXServerUtils;
+import org.apache.cassandra.utils.MBeanWrapper;
+import org.apache.cassandra.utils.RMIClientSocketFactoryImpl;
+import org.apache.cassandra.utils.ReflectionUtils;
+import sun.rmi.transport.tcp.TCPEndpoint;
+
+import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_RMI_DGC_LEASE_VALUE_IN_JVM_DTEST;
+import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION;
+import static org.apache.cassandra.config.CassandraRelevantProperties.SUN_RMI_TRANSPORT_TCP_THREADKEEPALIVETIME;
+import static org.apache.cassandra.distributed.api.Feature.JMX;
+
+public class IsolatedJmx
+{
+    private static final int RMI_KEEPALIVE_TIME = 1000;
+
+    private JMXConnectorServer jmxConnectorServer;
+    private JMXServerUtils.JmxRegistry registry;
+    private RMIJRMPServerImpl jmxRmiServer;
+    private MBeanWrapper.InstanceMBeanWrapper wrapper;
+    private RMIClientSocketFactoryImpl clientSocketFactory;
+    private CollectingRMIServerSocketFactoryImpl serverSocketFactory;
+    private Logger inInstancelogger;
+    private IInstanceConfig config;
+
+    public IsolatedJmx(IInstance instance, Logger inInstanceLogger) {
+        this.config = instance.config();
+        this.inInstancelogger = inInstanceLogger;
+    }
+
+    public void startJmx() {
+        try
+        {
+            // Several RMI threads hold references to in-jvm dtest objects, and are, by default, kept
+            // alive for long enough (minutes) to keep classloaders from being collected.
+            // Set these two system properties to a low value to allow cleanup to occur fast enough
+            // for GC to collect our classloaders.
+            JAVA_RMI_DGC_LEASE_VALUE_IN_JVM_DTEST.setInt(RMI_KEEPALIVE_TIME);
+            SUN_RMI_TRANSPORT_TCP_THREADKEEPALIVETIME.setInt(RMI_KEEPALIVE_TIME);
+            ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.setBoolean(false);
+            InetAddress addr = config.broadcastAddress().getAddress();
+
+            int jmxPort = config.jmxPort();
+
+            String hostname = addr.getHostAddress();
+            wrapper = new MBeanWrapper.InstanceMBeanWrapper(hostname + ":" + jmxPort);
+            ((MBeanWrapper.DelegatingMbeanWrapper) MBeanWrapper.instance).setDelegate(wrapper);
+            Map<String, Object> env = new HashMap<>();
+
+            serverSocketFactory = new CollectingRMIServerSocketFactoryImpl(addr);
+            env.put(RMIConnectorServer.RMI_SERVER_SOCKET_FACTORY_ATTRIBUTE,
+                    serverSocketFactory);
+            clientSocketFactory = new RMIClientSocketFactoryImpl(addr);
+            env.put(RMIConnectorServer.RMI_CLIENT_SOCKET_FACTORY_ATTRIBUTE,
+                    clientSocketFactory);
+
+            // configure the RMI registry
+            registry = new JMXServerUtils.JmxRegistry(jmxPort,
+                                                      clientSocketFactory,
+                                                      serverSocketFactory,
+                                                      "jmxrmi");
+
+            // Mark the JMX server as a permanently exported object. This allows the JVM to exit with the
+            // server running and also exempts it from the distributed GC scheduler which otherwise would
+            // potentially attempt a full GC every `sun.rmi.dgc.server.gcInterval` millis (default is 3600000ms)
+            // For more background see:
+            //   - CASSANDRA-2967
+            //   - https://www.jclarity.com/2015/01/27/rmi-system-gc-unplugged/
+            //   - https://bugs.openjdk.java.net/browse/JDK-6760712
+            env.put("jmx.remote.x.daemon", "true");
+
+            // Set the port used to create subsequent connections to exported objects over RMI. This simplifies
+            // configuration in firewalled environments, but it can't be used in conjuction with SSL sockets.
+            // See: CASSANDRA-7087
+            int rmiPort = config.jmxPort();
+
+            // We create the underlying RMIJRMPServerImpl so that we can manually bind it to the registry,
+            // rather then specifying a binding address in the JMXServiceURL and letting it be done automatically
+            // when the server is started. The reason for this is that if the registry is configured with SSL
+            // sockets, the JMXConnectorServer acts as its client during the binding which means it needs to
+            // have a truststore configured which contains the registry's certificate. Manually binding removes
+            // this problem.
+            // See CASSANDRA-12109.
+            jmxRmiServer = new RMIJRMPServerImpl(rmiPort, clientSocketFactory, serverSocketFactory,
+                                                 env);
+            JMXServiceURL serviceURL = new JMXServiceURL("rmi", hostname, rmiPort);
+            jmxConnectorServer = new RMIConnectorServer(serviceURL, env, jmxRmiServer, wrapper.getMBeanServer());
+
+            jmxConnectorServer.start();
+
+            registry.setRemoteServerStub(jmxRmiServer.toStub());
+            JMXServerUtils.logJmxServiceUrl(addr, jmxPort);
+            waitForJmxAvailability(hostname, jmxPort, env);
+        }
+        catch (Throwable t)
+        {
+            throw new RuntimeException("Feature.JMX was enabled but could not be started.", t);
+        }
+    }
+
+    private void waitForJmxAvailability(String hostname, int rmiPort, Map<String, Object> env) throws InterruptedException, MalformedURLException
+    {
+        String url = String.format("service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi", hostname, rmiPort);
+        JMXServiceURL serviceURL = new JMXServiceURL(url);
+        int attempts = 0;
+        Throwable lastThrown = null;
+        while (attempts < 20)
+        {
+            attempts++;
+            try (JMXConnector ignored = JMXConnectorFactory.connect(serviceURL, env))
+            {
+                inInstancelogger.info("Connected to JMX server at {} after {} attempt(s)",
+                                      url, attempts);
+                return;
+            }
+            catch (MalformedURLException e)
+            {
+                throw new RuntimeException(e);
+            }
+            catch (Throwable thrown)
+            {
+                lastThrown = thrown;
+            }
+            inInstancelogger.info("Could not connect to JMX on {} after {} attempts. Will retry.", url, attempts);
+            Thread.sleep(1000);
+        }
+        throw new RuntimeException("Could not start JMX - unreachable after 20 attempts", lastThrown);
+    }
+
+    public void stopJmx() throws IllegalAccessException, NoSuchFieldException, InterruptedException
+    {
+        if (!config.has(JMX))
+            return;
+        // First, swap the mbean wrapper back to a NoOp wrapper
+        // This prevents later attempts to unregister mbeans from failing in Cassandra code, as we're going to
+        // unregister all of them here
+        ((MBeanWrapper.DelegatingMbeanWrapper) MBeanWrapper.instance).setDelegate(new MBeanWrapper.NoOpMBeanWrapper());
+        try
+        {
+            wrapper.close();
+        }
+        catch (Throwable e)
+        {
+            inInstancelogger.warn("failed to close wrapper.", e);
+        }
+        try
+        {
+            jmxConnectorServer.stop();
+        }
+        catch (Throwable e)
+        {
+            inInstancelogger.warn("failed to close jmxConnectorServer.", e);
+        }
+        try
+        {
+            registry.close();
+        }
+        catch (Throwable e)
+        {
+            inInstancelogger.warn("failed to close registry.", e);
+        }
+        try
+        {
+            serverSocketFactory.close();
+        }
+        catch (Throwable e)
+        {
+            inInstancelogger.warn("failed to close serverSocketFactory.", e);
+        }
+        // The TCPEndpoint class holds references to a class in the in-jvm dtest framework
+        // which transitively has a reference to the InstanceClassLoader, so we need to
+        // make sure to remove the reference to them when the instance is shutting down
+        clearMapField(TCPEndpoint.class, null, "localEndpoints");
+        Thread.sleep(2 * RMI_KEEPALIVE_TIME); // Double the keep-alive time to give Distributed GC some time to clean up
+    }
+
+    private <K, V> void clearMapField(Class<?> clazz, Object instance, String mapName)
+    throws IllegalAccessException, NoSuchFieldException {
+        Field mapField = ReflectionUtils.getField(clazz, mapName);
+        mapField.setAccessible(true);
+        Map<K, V> map = (Map<K, V>) mapField.get(instance);
+        // Because multiple instances can be shutting down at once,
+        // synchronize on the map to avoid ConcurrentModificationException
+        synchronized (map)
+        {
+            for (Iterator<Map.Entry<K, V>> it = map.entrySet().iterator(); it.hasNext(); )
+            {
+                it.next();
+                it.remove();
+            }
+        }
+    }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java b/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java
index a5a2dce..794873f 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java
@@ -26,21 +26,29 @@
 import java.text.SimpleDateFormat;
 import java.util.function.Consumer;
 import javax.management.MBeanServer;
+import javax.management.MBeanServerConnection;
+import javax.management.remote.JMXConnector;
 
-import org.apache.cassandra.io.util.File;
+import org.junit.Assert;
 import org.junit.Ignore;
 import org.junit.Test;
 
 import com.sun.management.HotSpotDiagnosticMXBean;
 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.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.shared.JMXUtil;
+import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.utils.SigarLibrary;
 
 import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.JMX;
 import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
 import static org.apache.cassandra.distributed.api.Feature.NETWORK;
 import static org.apache.cassandra.utils.FBUtilities.now;
+import static org.hamcrest.Matchers.startsWith;
 
 /* Resource Leak Test - useful when tracking down issues with in-JVM framework cleanup.
  * All objects referencing the InstanceClassLoader need to be garbage collected or
@@ -139,6 +147,11 @@
 
     void doTest(int numClusterNodes, Consumer<IInstanceConfig> updater) throws Throwable
     {
+        doTest(numClusterNodes, updater, ignored -> {});
+    }
+
+    void doTest(int numClusterNodes, Consumer<IInstanceConfig> updater, Consumer<Cluster> actionToPerform) throws Throwable
+    {
         for (int loop = 0; loop < numTestLoops; loop++)
         {
             System.out.println(String.format("========== Starting loop %03d ========", loop));
@@ -149,6 +162,7 @@
                 cluster.schemaChange("CREATE TABLE " + KEYSPACE + "." + tableName + " (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
                 cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + "." + tableName + "(pk,ck,v) VALUES (0,0,0)", ConsistencyLevel.ALL);
                 cluster.get(1).flush(KEYSPACE);
+                actionToPerform.accept(cluster);
                 if (dumpEveryLoop)
                 {
                     dumpResources(String.format("loop%03d", loop));
@@ -207,4 +221,50 @@
         }
         dumpResources("final-native");
     }
+
+    @Test
+    public void looperJmxTest() throws Throwable
+    {
+        doTest(1, config -> config.with(JMX), cluster -> {
+            // NOTE: At some point, the hostname of the broadcastAddress can be resolved
+            // and then the `getHostString`, which would otherwise return the IP address,
+            // starts returning `localhost` - use `.getAddress().getHostAddress()` to work around this.
+            for (IInvokableInstance instance:cluster.get(1, cluster.size()))
+            {
+                IInstanceConfig config = instance.config();
+                try (JMXConnector jmxc = JMXUtil.getJmxConnector(config))
+                {
+                    MBeanServerConnection mbsc = jmxc.getMBeanServerConnection();
+                    // instances get their default domain set to their IP address, so us it
+                    // to check that we are actually connecting to the correct instance
+                    String defaultDomain = mbsc.getDefaultDomain();
+                    Assert.assertThat(defaultDomain, startsWith(JMXUtil.getJmxHost(config) + ":" + config.jmxPort()));
+                }
+                catch (IOException e)
+                {
+                    throw new RuntimeException(e);
+                }
+            }
+        });
+        if (forceCollection)
+        {
+            System.runFinalization();
+            System.gc();
+            Thread.sleep(finalWaitMillis);
+        }
+        dumpResources("final-jmx");
+    }
+
+    @Test
+    public void looperEverythingTest() throws Throwable
+    {
+        doTest(1, config -> config.with(Feature.values()));
+        if (forceCollection)
+        {
+            System.runFinalization();
+            System.gc();
+            Thread.sleep(finalWaitMillis);
+        }
+        dumpResources("final-everything");
+    }
 }
diff --git a/test/distributed/org/apache/cassandra/distributed/test/VirtualTableLogsTest.java b/test/distributed/org/apache/cassandra/distributed/test/VirtualTableLogsTest.java
index 1fc2958..71ef4db 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/VirtualTableLogsTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/VirtualTableLogsTest.java
@@ -27,6 +27,7 @@
 import org.apache.cassandra.db.virtual.LogMessagesTable;
 import org.apache.cassandra.db.virtual.LogMessagesTable.LogMessage;
 import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.Constants;
 import org.apache.cassandra.distributed.api.Feature;
 import org.apache.cassandra.distributed.api.SimpleQueryResult;
 import org.apache.cassandra.distributed.shared.WithProperties;
@@ -68,12 +69,17 @@
     public void testMultipleAppendersFailToStartNode() throws Throwable
     {
         LOGBACK_CONFIGURATION_FILE.setString("test/conf/logback-dtest_with_vtable_appender_invalid.xml");
+
+        // NOTE: Because cluster startup is expected to fail in this case, and can leave things in a weird state
+        // for the next state, create without starting, and set failure as shutdown to false,
+        // so the try-with-resources can close instances properly.
         try (WithProperties properties = new WithProperties().set(LOGBACK_CONFIGURATION_FILE, "test/conf/logback-dtest_with_vtable_appender_invalid.xml");
-             Cluster ignored = Cluster.build(1)
-                                      .withConfig(c -> c.with(Feature.values()))
-                                      .start();
-             )
+             Cluster cluster = Cluster.build(1)
+                                      .withConfig(c -> c.with(Feature.values())
+                                                        .set(Constants.KEY_DTEST_API_STARTUP_FAILURE_AS_SHUTDOWN, false))
+                                      .createWithoutStarting())
         {
+            cluster.startup();
             fail("Node should not start as there is supposed to be invalid logback configuration file.");
         }
         catch (IllegalStateException ex)
diff --git a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java
new file mode 100644
index 0000000..83a35e5
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.jmx;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import javax.management.MBeanServerConnection;
+import javax.management.remote.JMXConnector;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.impl.INodeProvisionStrategy;
+import org.apache.cassandra.distributed.shared.JMXUtil;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+
+import static org.hamcrest.Matchers.startsWith;
+
+public class JMXFeatureTest extends TestBaseImpl
+{
+
+    /**
+     * Test the in-jvm dtest JMX feature.
+     * - Create a cluster with multiple JMX servers, one per instance
+     * - Test that when connecting, we get the correct MBeanServer by checking the default domain, which is set to the IP of the instance
+     * - Run the test multiple times to ensure cleanup of the JMX servers is complete so the next test can run successfully using the same host/port.
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testMultipleNetworkInterfacesProvisioning() throws Exception
+    {
+        int iterations = 2; // Make sure the JMX infrastructure all cleans up properly by running this multiple times.
+        Set<String> allInstances = new HashSet<>();
+        for (int i = 0; i < iterations; i++)
+        {
+            try (Cluster cluster = Cluster.build(2)
+                                          .withNodeProvisionStrategy(INodeProvisionStrategy.Strategy.MultipleNetworkInterfaces)
+                                          .withConfig(c -> c.with(Feature.values())).start())
+            {
+                Set<String> instancesContacted = new HashSet<>();
+                for (IInvokableInstance instance : cluster.get(1, 2))
+                {
+                    testInstance(instancesContacted, instance);
+                }
+                Assert.assertEquals("Should have connected with both JMX instances.", 2, instancesContacted.size());
+                allInstances.addAll(instancesContacted);
+            }
+        }
+        Assert.assertEquals("Each instance from each cluster should have been unique", iterations * 2, allInstances.size());
+    }
+
+    @Test
+    public void testOneNetworkInterfaceProvisioning() throws Exception
+    {
+        int iterations = 2; // Make sure the JMX infrastructure all cleans up properly by running this multiple times.
+        Set<String> allInstances = new HashSet<>();
+        for (int i = 0; i < iterations; i++)
+        {
+            try (Cluster cluster = Cluster.build(2)
+                                          .withNodeProvisionStrategy(INodeProvisionStrategy.Strategy.OneNetworkInterface)
+                                          .withConfig(c -> c.with(Feature.values())).start())
+            {
+                Set<String> instancesContacted = new HashSet<>();
+                for (IInvokableInstance instance : cluster.get(1, 2))
+                {
+                    testInstance(instancesContacted, instance);
+                }
+                Assert.assertEquals("Should have connected with both JMX instances.", 2, instancesContacted.size());
+                allInstances.addAll(instancesContacted);
+            }
+        }
+        Assert.assertEquals("Each instance from each cluster should have been unique", iterations * 2, allInstances.size());
+    }
+
+    private void testInstance(Set<String> instancesContacted, IInvokableInstance instance) throws IOException
+    {
+        // NOTE: At some point, the hostname of the broadcastAddress can be resolved
+        // and then the `getHostString`, which would otherwise return the IP address,
+        // starts returning `localhost` - use `.getAddress().getHostAddress()` to work around this.
+        IInstanceConfig config = instance.config();
+        try (JMXConnector jmxc = JMXUtil.getJmxConnector(config))
+        {
+            MBeanServerConnection mbsc = jmxc.getMBeanServerConnection();
+            // instances get their default domain set to their IP address, so us it
+            // to check that we are actually connecting to the correct instance
+            String defaultDomain = mbsc.getDefaultDomain();
+            instancesContacted.add(defaultDomain);
+            Assert.assertThat(defaultDomain, startsWith(JMXUtil.getJmxHost(config) + ":" + config.jmxPort()));
+        }
+    }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java
index b935e24..ece6052 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java
@@ -17,7 +17,6 @@
  */
 package org.apache.cassandra.distributed.test.jmx;
 
-import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -31,7 +30,6 @@
 import javax.management.ObjectName;
 import javax.management.remote.JMXConnector;
 import javax.management.remote.JMXConnectorFactory;
-import javax.management.remote.JMXConnectorServer;
 import javax.management.remote.JMXServiceURL;
 
 import com.google.common.collect.ImmutableSet;
@@ -39,11 +37,8 @@
 
 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.test.TestBaseImpl;
-import org.apache.cassandra.utils.JMXServerUtils;
-
-import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION;
-import static org.apache.cassandra.cql3.CQLTester.getAutomaticallyAllocatedPort;
 
 public class JMXGetterCheckTest extends TestBaseImpl
 {
@@ -57,20 +52,17 @@
     "org.apache.cassandra.db:type=StorageService:resetLocalSchema" // this will fail when there are no other nodes which can serve schema
     );
 
-    @Test
-    public void test() throws Exception
-    {
-        // start JMX server, which the instance will register with
-        InetAddress loopback = InetAddress.getLoopbackAddress();
-        String jmxHost = loopback.getHostAddress();
-        int jmxPort = getAutomaticallyAllocatedPort(loopback);
-        JMXConnectorServer jmxServer = JMXServerUtils.createJMXServer(jmxPort, true);
-        jmxServer.start();
-        String url = "service:jmx:rmi:///jndi/rmi://" + jmxHost + ":" + jmxPort + "/jmxrmi";
+    public static final String JMX_SERVICE_URL_FMT = "service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi";
 
-        ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.setBoolean(false);
+    @Test
+    public void testGetters() throws Exception
+    {
         try (Cluster cluster = Cluster.build(1).withConfig(c -> c.with(Feature.values())).start())
         {
+            IInvokableInstance instance = cluster.get(1);
+
+            String jmxHost = instance.config().broadcastAddress().getAddress().getHostAddress();
+            String url = String.format(JMX_SERVICE_URL_FMT, jmxHost, instance.config().jmxPort());
             List<Named> errors = new ArrayList<>();
             try (JMXConnector jmxc = JMXConnectorFactory.connect(new JMXServiceURL(url), null))
             {
@@ -122,7 +114,7 @@
     }
 
     /**
-     * This class is meant to make new errors easier to read, by adding the JMX endpoint, and cleaning up the unneded JMX/Reflection logic cluttering the stacktrace
+     * This class is meant to make new errors easier to read, by adding the JMX endpoint, and cleaning up the unneeded JMX/Reflection logic cluttering the stacktrace
      */
     private static class Named extends RuntimeException
     {
diff --git a/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java b/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java
index a8ddfe0..58a3fee 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java
@@ -29,7 +29,9 @@
 import java.util.stream.Collectors;
 import javax.management.InstanceAlreadyExistsException;
 import javax.management.InstanceNotFoundException;
+import javax.management.MBeanServer;
 import javax.management.ObjectName;
+import javax.management.QueryExp;
 
 import com.google.common.collect.ImmutableSet;
 import org.junit.Assert;
@@ -57,7 +59,6 @@
         MBEAN_REGISTRATION_CLASS.setString(MapMBeanWrapper.class.getName());
         ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.setBoolean(false);
     }
-
     private static volatile Map<String, Collection<String>> SYSTEM_TABLES = null;
     private static Set<String> TABLE_METRIC_NAMES = ImmutableSet.of("WriteLatency");
 
@@ -165,7 +166,7 @@
     {
         inst.runOnInstance(() -> {
             // cast only to make sure it linked properly
-            MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance;
+            MapMBeanWrapper mbeans = getMapMBeanWrapper();
             Assert.assertTrue("Unable to find table mbean for " + keyspace + "." + table,
                               mbeans.isRegistered(ColumnFamilyStore.getTableMBeanName(keyspace, table, false)));
             Assert.assertTrue("Unable to find column family mbean for " + keyspace + "." + table,
@@ -177,7 +178,7 @@
     {
         inst.runOnInstance(() -> {
             // cast only to make sure it linked properly
-            MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance;
+            MapMBeanWrapper mbeans = getMapMBeanWrapper();
             Assert.assertFalse("Found table mbean for " + keyspace + "." + table,
                                mbeans.isRegistered(ColumnFamilyStore.getTableMBeanName(keyspace, table, false)));
             Assert.assertFalse("Found column family mbean for " + keyspace + "." + table,
@@ -189,7 +190,7 @@
     {
         inst.runOnInstance(() -> {
             // cast only to make sure it linked properly
-            MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance;
+            MapMBeanWrapper mbeans = getMapMBeanWrapper();
             String mbean = getTableMetricName(keyspace, table, name);
             Assert.assertTrue("Unable to find metric " + name + " for " + keyspace + "." + table, mbeans.isRegistered(mbean));
 
@@ -203,7 +204,7 @@
     {
         inst.runOnInstance(() -> {
             // cast only to make sure it linked properly
-            MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance;
+            MapMBeanWrapper mbeans = getMapMBeanWrapper();
             String mbean = getTableMetricName(keyspace, table, name);
             Assert.assertFalse("Found metric " + name + " for " + keyspace + "." + table, mbeans.isRegistered(mbean));
 
@@ -226,7 +227,7 @@
     {
         inst.runOnInstance(() -> {
             // cast only to make sure it linked properly
-            MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance;
+            MapMBeanWrapper mbeans = getMapMBeanWrapper();
 
             String keyspaceMBean = getKeyspaceMetricName(keyspace, name);
             Assert.assertFalse("Found keyspace metric " + keyspaceMBean + " for " + keyspace, mbeans.isRegistered(keyspaceMBean));
@@ -243,6 +244,10 @@
         return String.format("org.apache.cassandra.metrics:type=Table,keyspace=%s,scope=%s,name=%s", keyspace, table, name);
     }
 
+    private static MapMBeanWrapper getMapMBeanWrapper()
+    {
+        return (MapMBeanWrapper) ((MBeanWrapper.DelegatingMbeanWrapper)MBeanWrapper.instance).getDelegate();
+    }
     public static final class MapMBeanWrapper implements MBeanWrapper
     {
         private final ConcurrentMap<ObjectName, Object> map = new ConcurrentHashMap<>();
@@ -268,5 +273,17 @@
             if (previous == null)
                 onException.handler.accept(new InstanceNotFoundException("MBean " + mbeanName + " was not found"));
         }
+
+        @Override
+        public Set<ObjectName> queryNames(ObjectName name, QueryExp query)
+        {
+            return null;
+        }
+
+        @Override
+        public MBeanServer getMBeanServer()
+        {
+            return null;
+        }
     }
 }