diff --git a/pymongo/periodic_executor.py b/pymongo/periodic_executor.py
index ba9664fa7..5777e5ab2 100644
--- a/pymongo/periodic_executor.py
+++ b/pymongo/periodic_executor.py
@@ -50,6 +50,10 @@ class PeriodicExecutor(object):
         self._thread_will_exit = False
         self._lock = threading.Lock()
 
+    def __repr__(self):
+        return '<%s(name=%s) object at 0x%x>' % (
+            self.__class__.__name__, self._name, id(self))
+
     def open(self):
         """Start. Multiple calls have no effect.
 
diff --git a/pymongo/settings.py b/pymongo/settings.py
index 2a02f05d5..dd0ac3c1a 100644
--- a/pymongo/settings.py
+++ b/pymongo/settings.py
@@ -15,6 +15,7 @@
 """Represent MongoClient's configuration."""
 
 import threading
+import traceback
 
 from bson.objectid import ObjectId
 from pymongo import common, monitor, pool
@@ -60,6 +61,9 @@ class TopologySettings(object):
         self._heartbeat_frequency = heartbeat_frequency
         self._direct = (len(self._seeds) == 1 and not replica_set_name)
         self._topology_id = ObjectId()
+        # Store the allocation traceback to catch unclosed clients in the
+        # test suite.
+        self._stack = ''.join(traceback.format_stack())
 
     @property
     def seeds(self):
diff --git a/test/__init__.py b/test/__init__.py
index a82e62edf..73a91c0b2 100644
--- a/test/__init__.py
+++ b/test/__init__.py
@@ -795,6 +795,44 @@ def setup():
     warnings.simplefilter("always")
 
 
+def _get_executors(topology):
+    executors = []
+    for server in topology._servers.values():
+        # Some MockMonitor do not have an _executor.
+        executors.append(getattr(server._monitor, '_executor', None))
+    executors.append(topology._Topology__events_executor)
+    if topology._srv_monitor:
+        executors.append(topology._srv_monitor._executor)
+    return [e for e in executors if e is not None]
+
+
+def all_executors_stopped(topology):
+    running = [e for e in _get_executors(topology) if not e._stopped]
+    if running:
+        print('  Topology %s has THREADS RUNNING: %s, created at: %s' % (
+            topology, running, topology._settings._stack))
+        return False
+    return True
+
+
+def print_unclosed_clients():
+    from pymongo.topology import Topology
+    processed = set()
+    # Call collect to manually cleanup any would-be gc'd clients to avoid
+    # false positives.
+    gc.collect()
+    for obj in gc.get_objects():
+        try:
+            if isinstance(obj, Topology):
+                # Avoid printing the same Topology multiple times.
+                if obj._topology_id in processed:
+                    continue
+                all_executors_stopped(obj)
+                processed.add(obj._topology_id)
+        except ReferenceError:
+            pass
+
+
 def teardown():
     garbage = []
     for g in gc.garbage:
@@ -813,6 +851,10 @@ def teardown():
         c.drop_database("pymongo_test_bernie")
         c.close()
 
+    # Jython does not support gc.get_objects.
+    if not sys.platform.startswith('java'):
+        print_unclosed_clients()
+
 
 class PymongoTestRunner(unittest.TextTestRunner):
     def run(self, test):
diff --git a/test/pymongo_mocks.py b/test/pymongo_mocks.py
index bd3e1ae22..388f89178 100644
--- a/test/pymongo_mocks.py
+++ b/test/pymongo_mocks.py
@@ -65,8 +65,9 @@ class MockMonitor(Monitor):
             topology,
             pool,
             topology_settings):
-        # MockMonitor gets a 'client' arg, regular monitors don't.
-        self.client = client
+        # MockMonitor gets a 'client' arg, regular monitors don't. Weakref it
+        # to avoid cycles.
+        self.client = weakref.proxy(client)
         Monitor.__init__(
             self,
             server_description,
@@ -75,8 +76,9 @@ class MockMonitor(Monitor):
             topology_settings)
 
     def _check_once(self):
+        client = self.client
         address = self._server_description.address
-        response, rtt = self.client.mock_is_master('%s:%d' % address)
+        response, rtt = client.mock_is_master('%s:%d' % address)
         return ServerDescription(address, IsMaster(response), rtt)
 
 
diff --git a/test/test_auth.py b/test/test_auth.py
index 6dccc6ff1..c8e4ef1ab 100644
--- a/test/test_auth.py
+++ b/test/test_auth.py
@@ -695,8 +695,6 @@ class TestAuthURIOptions(unittest.TestCase):
         client_context.create_user('admin', 'admin', 'pass')
         client_context.create_user(
             'pymongo_test', 'user', 'pass', ['userAdmin', 'readWrite'])
-        self.client = rs_or_single_client_noauth(
-            username='admin', password='pass')
 
     def tearDown(self):
         client_context.drop_user('pymongo_test', 'user')
diff --git a/test/test_client.py b/test/test_client.py
index ef1b19485..df2221eb3 100644
--- a/test/test_client.py
+++ b/test/test_client.py
@@ -101,6 +101,10 @@ class ClientUnitTest(unittest.TestCase):
         cls.client = rs_or_single_client(connect=False,
                                          serverSelectionTimeoutMS=100)
 
+    @classmethod
+    def tearDownClass(cls):
+        cls.client.close()
+
     def test_keyword_arg_defaults(self):
         client = MongoClient(socketTimeoutMS=None,
                              connectTimeoutMS=20000,
diff --git a/test/test_collation.py b/test/test_collation.py
index 7cb4d8b5c..d87a2a9ac 100644
--- a/test/test_collation.py
+++ b/test/test_collation.py
@@ -105,6 +105,7 @@ class TestCollation(unittest.TestCase):
     def tearDownClass(cls):
         cls.warn_context.__exit__()
         cls.warn_context = None
+        cls.client.close()
 
     def tearDown(self):
         self.listener.results.clear()
diff --git a/test/test_command_monitoring_spec.py b/test/test_command_monitoring_spec.py
index 3d41d4b48..3363a4256 100644
--- a/test/test_command_monitoring_spec.py
+++ b/test/test_command_monitoring_spec.py
@@ -48,6 +48,10 @@ class TestAllScenarios(unittest.TestCase):
         cls.listener = EventListener()
         cls.client = single_client(event_listeners=[cls.listener])
 
+    @classmethod
+    def tearDownClass(cls):
+        cls.client.close()
+
     def tearDown(self):
         self.listener.results.clear()
 
diff --git a/test/test_connections_survive_primary_stepdown_spec.py b/test/test_connections_survive_primary_stepdown_spec.py
index 4a63e0e23..63cf127e3 100644
--- a/test/test_connections_survive_primary_stepdown_spec.py
+++ b/test/test_connections_survive_primary_stepdown_spec.py
@@ -51,6 +51,10 @@ class TestConnectionsSurvivePrimaryStepDown(IntegrationTest):
         cls.coll = cls.db.get_collection(
             "step-down", write_concern=WriteConcern("majority"))
 
+    @classmethod
+    def tearDownClass(cls):
+        cls.client.close()
+
     def setUp(self):
         # Note that all ops use same write-concern as self.db (majority).
         self.db.drop_collection("step-down")
diff --git a/test/test_custom_types.py b/test/test_custom_types.py
index ba0bb0ca6..41b79a96d 100644
--- a/test/test_custom_types.py
+++ b/test/test_custom_types.py
@@ -911,6 +911,7 @@ class TestClusterChangeStreamsWCustomTypes(
             kwargs['type_registry'] = codec_options.type_registry
             kwargs['document_class'] = codec_options.document_class
         self.watched_target = rs_client(*args, **kwargs)
+        self.addCleanup(self.watched_target.close)
         self.input_target = self.watched_target[self.db.name].test
         # Insert a record to ensure db, coll are created.
         self.input_target.insert_one({'data': 'dummy'})
diff --git a/test/test_examples.py b/test/test_examples.py
index f2747ff46..16e1936d5 100644
--- a/test/test_examples.py
+++ b/test/test_examples.py
@@ -27,27 +27,26 @@ from pymongo.read_preferences import ReadPreference
 from pymongo.write_concern import WriteConcern
 
 from test import client_context, unittest, IntegrationTest
-from test.utils import rs_client, rs_or_single_client
+from test.utils import rs_client
 
-class TestSampleShellCommands(unittest.TestCase):
 
+class TestSampleShellCommands(IntegrationTest):
     @classmethod
-    @client_context.require_connection
     def setUpClass(cls):
-        cls.client = rs_or_single_client(w="majority")
+        super(TestSampleShellCommands, cls).setUpClass()
         # Run once before any tests run.
-        cls.client.pymongo_test.inventory.drop()
+        cls.db.inventory.drop()
 
     @classmethod
     def tearDownClass(cls):
-        client_context.client.drop_database("pymongo_test")
+        cls.client.drop_database("pymongo_test")
 
     def tearDown(self):
         # Run after every test.
-        self.client.pymongo_test.inventory.drop()
+        self.db.inventory.drop()
 
     def test_first_three_examples(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Example 1
         db.inventory.insert_one(
@@ -84,7 +83,7 @@ class TestSampleShellCommands(unittest.TestCase):
         self.assertEqual(db.inventory.count_documents({}), 4)
 
     def test_query_top_level_fields(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Example 6
         db.inventory.insert_many([
@@ -151,7 +150,7 @@ class TestSampleShellCommands(unittest.TestCase):
         self.assertEqual(len(list(cursor)), 2)
 
     def test_query_embedded_documents(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Example 14
         # Subdocument key order matters in a few of these examples so we have
@@ -214,7 +213,7 @@ class TestSampleShellCommands(unittest.TestCase):
         self.assertEqual(len(list(cursor)), 1)
 
     def test_query_arrays(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Example 20
         db.inventory.insert_many([
@@ -290,7 +289,7 @@ class TestSampleShellCommands(unittest.TestCase):
         self.assertEqual(len(list(cursor)), 1)
 
     def test_query_array_of_documents(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Example 29
         # Subdocument key order matters in a few of these examples so we have
@@ -372,7 +371,7 @@ class TestSampleShellCommands(unittest.TestCase):
         self.assertEqual(len(list(cursor)), 2)
 
     def test_query_null(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Example 38
         db.inventory.insert_many([{"_id": 1, "item": None}, {"_id": 2}])
@@ -397,7 +396,7 @@ class TestSampleShellCommands(unittest.TestCase):
         self.assertEqual(len(list(cursor)), 1)
 
     def test_projection(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Example 42
         db.inventory.insert_many([
@@ -528,7 +527,7 @@ class TestSampleShellCommands(unittest.TestCase):
             self.assertEqual(len(doc["instock"]), 1)
 
     def test_update_and_replace(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Example 51
         db.inventory.insert_many([
@@ -614,7 +613,7 @@ class TestSampleShellCommands(unittest.TestCase):
             self.assertEqual(len(doc["instock"]), 2)
 
     def test_delete(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Example 55
         db.inventory.insert_many([
@@ -664,7 +663,7 @@ class TestSampleShellCommands(unittest.TestCase):
     @client_context.require_replica_set
     @client_context.require_no_mmap
     def test_change_streams(self):
-        db = client_context.client.pymongo_test
+        db = self.db
         done = False
 
         def insert_docs():
@@ -706,7 +705,7 @@ class TestSampleShellCommands(unittest.TestCase):
             t.join()
 
     def test_aggregate_examples(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Aggregation Example 1
         db.sales.aggregate([
@@ -792,7 +791,7 @@ class TestSampleShellCommands(unittest.TestCase):
             # End Aggregation Example 4
 
     def test_commands(self):
-        db = client_context.client.pymongo_test
+        db = self.db
         db.restaurants.insert_one({})
 
         # Start runCommand Example 1
@@ -804,7 +803,7 @@ class TestSampleShellCommands(unittest.TestCase):
         # End runCommand Example 2
 
     def test_index_management(self):
-        db = client_context.client.pymongo_test
+        db = self.db
 
         # Start Index Example 1
         db.records.create_index("score")
@@ -821,7 +820,7 @@ class TestSampleShellCommands(unittest.TestCase):
     @client_context.require_replica_set
     def test_misc(self):
         # Marketing examples
-        client = client_context.client
+        client = self.client
         self.addCleanup(client.drop_database, "test")
         self.addCleanup(client.drop_database, "my_database")
 
@@ -843,13 +842,6 @@ class TestSampleShellCommands(unittest.TestCase):
 
 
 class TestTransactionExamples(IntegrationTest):
-
-    @classmethod
-    @client_context.require_connection
-    def setUpClass(cls):
-        super(TestTransactionExamples, cls).setUpClass()
-        cls.client = rs_or_single_client(w="majority")
-
     @client_context.require_version_max(4, 4, 99)  # PYTHON-2154 skip on 4.5+
     @client_context.require_transactions
     def test_transactions(self):
diff --git a/test/test_legacy_api.py b/test/test_legacy_api.py
index 9e63e59a8..fb1bd2b8a 100644
--- a/test/test_legacy_api.py
+++ b/test/test_legacy_api.py
@@ -2306,6 +2306,8 @@ class TestLegacyBulkWriteConcern(BulkTestBase):
     @classmethod
     def tearDownClass(cls):
         cls.deprecation_filter.stop()
+        if cls.secondary:
+            cls.secondary.close()
 
     def cause_wtimeout(self, batch):
         if self.need_replication_stopped:
diff --git a/test/test_mongos_load_balancing.py b/test/test_mongos_load_balancing.py
index 7df2239e7..18e05125b 100644
--- a/test/test_mongos_load_balancing.py
+++ b/test/test_mongos_load_balancing.py
@@ -71,6 +71,7 @@ class TestMongosLoadBalancing(MockClientTest):
             host='a:1,b:2,c:3',
             connect=False,
             **kwargs)
+        self.addCleanup(mock_client.close)
 
         # Latencies in seconds.
         mock_client.mock_rtts['a:1'] = 0.020
diff --git a/test/test_monitoring.py b/test/test_monitoring.py
index 52f9d121c..ede08c011 100644
--- a/test/test_monitoring.py
+++ b/test/test_monitoring.py
@@ -52,6 +52,10 @@ class TestCommandMonitoring(PyMongoTestCase):
             event_listeners=[cls.listener],
             retryWrites=False)
 
+    @classmethod
+    def tearDownClass(cls):
+        cls.client.close()
+
     def tearDown(self):
         self.listener.results.clear()
 
@@ -1401,6 +1405,7 @@ class TestGlobalListener(PyMongoTestCase):
     @classmethod
     def tearDownClass(cls):
         monitoring._LISTENERS = cls.saved_listeners
+        cls.client.close()
 
     def setUp(self):
         self.listener.results.clear()
diff --git a/test/test_pooling.py b/test/test_pooling.py
index 8ed4068a6..b1728d791 100644
--- a/test/test_pooling.py
+++ b/test/test_pooling.py
@@ -161,6 +161,9 @@ class _TestPoolingBase(unittest.TestCase):
         db.unique.insert_one({"_id": "jesse"})
         db.test.insert_many([{} for _ in range(10)])
 
+    def tearDown(self):
+        self.c.close()
+
     def create_pool(
             self,
             pair=(client_context.host, client_context.port),
diff --git a/test/test_read_concern.py b/test/test_read_concern.py
index abd69309a..2eef4cb1d 100644
--- a/test/test_read_concern.py
+++ b/test/test_read_concern.py
@@ -34,6 +34,7 @@ class TestReadConcern(PyMongoTestCase):
 
     @classmethod
     def tearDownClass(cls):
+        cls.client.close()
         client_context.client.pymongo_test.drop_collection('coll')
 
     def tearDown(self):
diff --git a/test/test_read_preferences.py b/test/test_read_preferences.py
index ce79592bb..821c277e8 100644
--- a/test/test_read_preferences.py
+++ b/test/test_read_preferences.py
@@ -370,6 +370,7 @@ class TestCommandAndReadPreference(TestReplicaSetClientBase):
     @classmethod
     def tearDownClass(cls):
         cls.c.drop_database('pymongo_test')
+        cls.c.close()
 
     def executed_on_which_server(self, client, fn, *args, **kwargs):
         """Execute fn(*args, **kwargs) and return the Server instance used."""
diff --git a/test/test_replica_set_client.py b/test/test_replica_set_client.py
index e0456186a..c65378418 100644
--- a/test/test_replica_set_client.py
+++ b/test/test_replica_set_client.py
@@ -300,6 +300,7 @@ class TestReplicaSetWireVersion(MockClientTest):
             host='a:1',
             replicaSet='rs',
             connect=False)
+        self.addCleanup(c.close)
 
         c.set_wire_version_range('a:1', 3, 7)
         c.set_wire_version_range('b:2', 2, 3)
@@ -330,15 +331,17 @@ class TestReplicaSetClientInternalIPs(MockClientTest):
     def test_connect_with_internal_ips(self):
         # Client is passed an IP it can reach, 'a:1', but the RS config
         # only contains unreachable IPs like 'internal-ip'. PYTHON-608.
+        client = MockClient(
+            standalones=[],
+            members=['a:1'],
+            mongoses=[],
+            ismaster_hosts=['internal-ip:27017'],
+            host='a:1',
+            replicaSet='rs',
+            serverSelectionTimeoutMS=100)
+        self.addCleanup(client.close)
         with self.assertRaises(AutoReconnect) as context:
-            connected(MockClient(
-                standalones=[],
-                members=['a:1'],
-                mongoses=[],
-                ismaster_hosts=['internal-ip:27017'],
-                host='a:1',
-                replicaSet='rs',
-                serverSelectionTimeoutMS=100))
+            connected(client)
 
         self.assertEqual(
             "Could not reach any servers in [('internal-ip', 27017)]."
@@ -356,6 +359,7 @@ class TestReplicaSetClientMaxWriteBatchSize(MockClientTest):
             host='a:1',
             replicaSet='rs',
             connect=False)
+        self.addCleanup(c.close)
 
         c.set_max_write_batch_size('a:1', 1)
         c.set_max_write_batch_size('b:2', 2)
diff --git a/test/test_retryable_writes.py b/test/test_retryable_writes.py
index 7f9a429b1..88a122d51 100644
--- a/test/test_retryable_writes.py
+++ b/test/test_retryable_writes.py
@@ -192,6 +192,7 @@ class TestRetryableWritesMMAPv1(IgnoreDeprecationsTest):
     @classmethod
     def tearDownClass(cls):
         cls.knobs.disable()
+        cls.client.close()
 
     @client_context.require_version_min(3, 5)
     @client_context.require_no_standalone
@@ -226,6 +227,7 @@ class TestRetryableWrites(IgnoreDeprecationsTest):
     @classmethod
     def tearDownClass(cls):
         cls.knobs.disable()
+        cls.client.close()
         super(TestRetryableWrites, cls).tearDownClass()
 
     def setUp(self):
diff --git a/test/test_session.py b/test/test_session.py
index 3c90fa5fa..a7c8e54e2 100644
--- a/test/test_session.py
+++ b/test/test_session.py
@@ -78,6 +78,7 @@ class TestSession(IntegrationTest):
     @classmethod
     def tearDownClass(cls):
         monitoring._SENSITIVE_COMMANDS.update(cls.sensitive_commands)
+        cls.client2.close()
         super(TestSession, cls).tearDownClass()
 
     def setUp(self):
@@ -85,6 +86,7 @@ class TestSession(IntegrationTest):
         self.session_checker_listener = SessionTestListener()
         self.client = rs_or_single_client(
             event_listeners=[self.listener, self.session_checker_listener])
+        self.addCleanup(self.client.close)
         self.db = self.client.pymongo_test
         self.initial_lsids = set(s['id'] for s in session_ids(self.client))
 
@@ -783,6 +785,10 @@ class TestCausalConsistency(unittest.TestCase):
         cls.listener = SessionTestListener()
         cls.client = rs_or_single_client(event_listeners=[cls.listener])
 
+    @classmethod
+    def tearDownClass(cls):
+        cls.client.close()
+
     @client_context.require_sessions
     def setUp(self):
         super(TestCausalConsistency, self).setUp()
diff --git a/test/test_transactions.py b/test/test_transactions.py
index cfcc67e95..ff707c5f9 100644
--- a/test/test_transactions.py
+++ b/test/test_transactions.py
@@ -56,6 +56,12 @@ class TransactionsBase(SpecRunner):
             for address in client_context.mongoses:
                 cls.mongos_clients.append(single_client('%s:%s' % address))
 
+    @classmethod
+    def tearDownClass(cls):
+        for client in cls.mongos_clients:
+            client.close()
+        super(TransactionsBase, cls).tearDownClass()
+
     def maybe_skip_scenario(self, test):
         super(TransactionsBase, self).maybe_skip_scenario(test)
         if ('secondary' in self.id() and