diff --git a/cassandra/cluster.py b/cassandra/cluster.py
index d858f5835e..96453d151f 100644
--- a/cassandra/cluster.py
+++ b/cassandra/cluster.py
@@ -3768,7 +3768,7 @@ class ControlConnection(object):
     _SELECT_PEERS = "SELECT peer, data_center, host_id, rack, release_version, rpc_address, schema_version, tokens FROM system.peers"
     _SELECT_PEERS_NO_TOKENS_TEMPLATE = "SELECT host_id, peer, data_center, rack, rpc_address, {nt_col_name}, release_version, schema_version FROM system.peers"
     _SELECT_LOCAL = "SELECT broadcast_address, cluster_name, data_center, host_id, listen_address, partitioner, rack, release_version, rpc_address, schema_version, tokens FROM system.local WHERE key='local'"
-    _SELECT_LOCAL_NO_TOKENS = "SELECT host_id, cluster_name, data_center, rack, partitioner, release_version, schema_version, rpc_address FROM system.local WHERE key='local'"
+    _SELECT_LOCAL_NO_TOKENS = "SELECT broadcast_address, cluster_name, data_center, host_id, listen_address, partitioner, rack, release_version, rpc_address, schema_version FROM system.local WHERE key='local'"
     # Used only when token_metadata_enabled is set to False
     _SELECT_LOCAL_NO_TOKENS_RPC_ADDRESS = "SELECT rpc_address FROM system.local WHERE key='local'"
 
@@ -4217,6 +4217,8 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
             host.dse_workloads = row.get("workloads")
 
             if row is local_row:
+                if "listen_address" in row:
+                    host.listen_address = row["listen_address"]
                 connection._control_connection_host_id = host_id
 
             tokens = row.get("tokens", None)
diff --git a/cassandra/pool.py b/cassandra/pool.py
index 2cd376d293..108926996e 100644
--- a/cassandra/pool.py
+++ b/cassandra/pool.py
@@ -98,9 +98,8 @@ class Host(object):
 
     'system.local.listen_address'
 
-    This is only available in the ``system.local`` table for newer versions of Cassandra. It is also not
-    queried if :attr:`~.Cluster.token_metadata_enabled` is ``False``. Usually the same as ``broadcast_address``
-    unless configured differently in cassandra.yaml.
+    This is only available in the ``system.local`` table for newer versions of Cassandra. Usually the
+    same as ``broadcast_address`` unless configured differently in cassandra.yaml.
     """
 
     listen_port = None
diff --git a/tests/integration/standard/test_metadata.py b/tests/integration/standard/test_metadata.py
index 562f457a32..04a042ebcc 100644
--- a/tests/integration/standard/test_metadata.py
+++ b/tests/integration/standard/test_metadata.py
@@ -73,17 +73,15 @@ def test_host_addresses(self):
                 assert host.broadcast_rpc_port is not None
 
         con = self.cluster.control_connection.get_connections()[0]
-        local_host = con.host
+        local_host = self.cluster.control_connection._get_host_for_connection(con)
 
         # The control connection node should have the listen address set.
-        # Note: Scylla does not populate listen_address in system.local
-        if SCYLLA_VERSION is None:
-            listen_addrs = [host.listen_address for host in self.cluster.metadata.all_hosts()]
-            assert local_host in listen_addrs
+        assert local_host is not None
+        assert local_host.listen_address is not None
 
         # The control connection node should have the broadcast_rpc_address set.
         rpc_addrs = [host.broadcast_rpc_address for host in self.cluster.metadata.all_hosts()]
-        assert local_host in rpc_addrs
+        assert con.host in rpc_addrs
 
     @unittest.skipUnless(
         os.getenv('MAPPED_CASSANDRA_VERSION', None) is not None,
diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py
index dec61dacdc..efcecc1f30 100644
--- a/tests/unit/test_control_connection.py
+++ b/tests/unit/test_control_connection.py
@@ -354,6 +354,37 @@ def test_refresh_nodes_and_tokens(self):
 
         assert self.connection.wait_for_responses.call_count == 1
 
+    def test_refresh_sets_local_listen_address_when_rpc_address_changes(self):
+        self.connection.local_results[0].append('listen_address')
+        self.connection.local_results[1][0].append('192.168.1.0')
+        self.connection.local_results[1][0][0] = '192.168.1.4'
+
+        self.control_connection.refresh_node_list_and_token_map()
+
+        local_host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        assert local_host.endpoint == DefaultEndPoint('192.168.1.4')
+        assert local_host.listen_address == '192.168.1.0'
+
+    def test_refresh_sets_local_addresses_without_token_metadata(self):
+        self.control_connection._token_meta_enabled = False
+        self.connection.local_results[0].append('listen_address')
+        self.connection.local_results[1][0].append('192.168.1.0')
+        self.connection.local_results[0].append('broadcast_address')
+        self.connection.local_results[1][0].append('10.0.0.1')
+
+        for results in (self.connection.local_results, self.connection.peer_results):
+            tokens_index = results[0].index('tokens')
+            results[0].pop(tokens_index)
+            for row in results[1]:
+                row.pop(tokens_index)
+        self.control_connection.refresh_node_list_and_token_map()
+
+        local_query = self.connection.wait_for_responses.call_args[0][1]
+        assert local_query.query == ControlConnection._SELECT_LOCAL_NO_TOKENS
+        local_host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        assert local_host.listen_address == '192.168.1.0'
+        assert local_host.broadcast_address == '10.0.0.1'
+
     def test_refresh_uses_control_endpoint_for_local_unix_host(self):
         maintenance_endpoint = UnixSocketEndPoint('/tmp/maintenance.sock')
         self._forget_local_host()