pool: make Host endpoint read-only

CHANGELOG.rst

@@ -11,6 +11,14 @@ Features
   ``Session.execute_concurrent_async()``, which returns a
   ``concurrent.futures.Future`` without blocking the caller.
 
+Compatibility Changes
+---------------------
+* ``Host.endpoint`` is now read-only, like ``Host.host_id``. Assigning or
+  deleting either property raises ``AttributeError``. Endpoint moves publish
+  a new ``Host`` with the same UUID and retire the previous object, whose
+  endpoint stays unchanged. Use cluster metadata to find the current host
+  instead of assigning an endpoint directly (#924).
+
 Bug Fixes
 ---------
 * Avoid loading optional NumPy during ordinary driver imports. NumPy is now

cassandra/pool.py

@@ -53,11 +53,6 @@ class Host(object):
     Represents a single Cassandra node.
     """
 
-    endpoint = None
-    """
-    The :class:`~.connection.EndPoint` to connect to the node.
-    """
-
     broadcast_address = None
     """
     broadcast address configured for the node, *if available*:
@@ -181,13 +176,23 @@ def __init__(self, endpoint, conviction_policy_factory, datacenter=None, rack=No
         if host_id.int == 0:
             raise ValueError("host_id may not be the nil UUID")
 
-        self.endpoint = endpoint if isinstance(endpoint, EndPoint) else DefaultEndPoint(endpoint)
+        self._endpoint = endpoint if isinstance(endpoint, EndPoint) else DefaultEndPoint(endpoint)
         self._host_id = host_id
         self._is_removed = False
         self.conviction_policy = conviction_policy_factory(self)
         self.set_location_info(datacenter, rack)
         self.lock = RLock()
 
+    @property
+    def endpoint(self):
+        """
+        The read-only :class:`~.connection.EndPoint` to connect to the node.
+
+        When the endpoint changes, the driver replaces this Host with a new
+        object carrying the same :attr:`host_id` and retires the old object.
+        """
+        return self._endpoint
+
     @property
     def address(self):
         """

docs/api/cassandra/pool.rst

@@ -5,6 +5,13 @@ Hosts and Connection Pools
 
 .. automodule:: cassandra.pool
 
+``Host.host_id`` and ``Host.endpoint`` are read-only. A node endpoint move
+publishes a distinct ``Host`` with the same UUID and retires the previous
+object. Retained references keep their original endpoint; consult cluster
+metadata for the current ``Host``. Equality and hashing use the UUID, so old
+and replacement objects compare equal even though they are distinct objects.
+Host health, conviction policy, and reconnection state remain mutable.
+
 .. autoclass:: Host ()
     :members:
     :exclude-members: set_location_info, get_and_set_reconnection_handler

tests/integration/standard/test_ip_change.py

@@ -3,6 +3,7 @@
 import unittest
 
 from cassandra.cluster import ExecutionProfile
+from cassandra.connection import DefaultEndPoint
 from cassandra.policies import WhiteListRoundRobinPolicy
 
 from tests.integration import use_cluster, get_node, get_cluster, local, TestCluster
@@ -39,6 +40,10 @@ def teardown_class(cls):
 
     def test_change_address_during_live_session(self):
         node3 = get_node(3)
+        old_endpoint = DefaultEndPoint(*node3.network_interfaces['binary'])
+        old_host = self.cluster.metadata.get_host(old_endpoint)
+        assert old_host is not None
+        host_id = old_host.host_id
 
         LOGGER.debug("Stop node3")
         node3.stop()
@@ -52,9 +57,16 @@ def test_change_address_during_live_session(self):
         node3.start(wait_for_binary_proto=True)
 
         def new_address_found():
-            addresses = [str(host.endpoint.address) for host in self.cluster.metadata.all_hosts()]
-            LOGGER.debug(addresses)
-            assert new_ip in addresses
+            metadata = self.cluster.metadata
+            current = metadata.get_host_by_host_id(host_id)
+            assert current is not None
+            assert current is not old_host
+            assert current.host_id == host_id
+            assert current.endpoint == DefaultEndPoint(new_ip, old_endpoint.port)
+            assert old_host.endpoint == old_endpoint
+            assert old_host._is_removed
+            assert metadata.get_host(old_endpoint) is None
+            assert metadata.get_host(current.endpoint) is current
 
         wait_until_not_raised(new_address_found, 0.5, 100)
 
@@ -66,5 +78,6 @@ def new_node_connectable():
             local_info = self.session.execute("SELECT * FROM system.local WHERE key='local'", execution_profile="new_node").one()
             LOGGER.debug(local_info._asdict())
             assert local_info.broadcast_address == new_ip
+            assert local_info.host_id == host_id
 
         wait_until_not_raised(new_node_connectable, 0.5, 100)

tests/unit/test_cluster.py

@@ -43,7 +43,7 @@
 from cassandra.pool import Host, _HostReconnectionHandler
 from cassandra.policies import ExponentialReconnectionPolicy, HostDistance, RetryPolicy, RoundRobinPolicy, DowngradingConsistencyRetryPolicy, SimpleConvictionPolicy
 from cassandra.query import SimpleStatement, named_tuple_factory, tuple_factory
-from tests.unit.utils import (mock_session_pools, new_session_with_pool_state,
+from tests.unit.utils import (mock_session_pools, move_host, new_session_with_pool_state,
                               run_isolated_subprocess)
 from tests import connection_class
 import pytest
@@ -3076,7 +3076,7 @@ def test_pool_creation_discards_result_after_endpoint_change(self):
         with patch('cassandra.cluster.HostConnection', side_effect=factory):
             future = session.add_or_renew_pool(host, is_host_addition=True)
             assert factory.creation_started.wait(5)
-            host.endpoint = DefaultEndPoint('127.0.0.2')
+            move_host(host, DefaultEndPoint('127.0.0.2'))
             factory.release_creation.set()
             assert future.result(timeout=5) is False
 
@@ -5200,7 +5200,7 @@ def test_set_keyspace_for_all_pools_reports_all_errors(self, *_):
         callback.assert_called_once()
         assert callback.call_args.args[0] == {'host1': [keyspace_error]}
 
-    def test_remove_pool_after_host_endpoint_changes(self):
+    def test_remove_pool_by_equal_host_at_another_endpoint(self):
         host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4())
         pool = Mock(host=host)
         shutdown_future = Future()
@@ -5209,7 +5209,6 @@ def test_remove_pool_after_host_endpoint_changes(self):
         session.cluster.executor.submit.return_value = shutdown_future
         session.is_shutdown = False
 
-        host.endpoint = DefaultEndPoint("127.0.0.2")
         same_host_at_another_endpoint = Host(
             "127.0.0.3", SimpleConvictionPolicy, host_id=host.host_id)
 

tests/unit/test_control_connection.py

@@ -1891,9 +1891,9 @@ def test_skipped_local_row_does_not_identify_endpoint_owner_as_control_host(self
 
         def move_owner_before_local_row(lookup_endpoint):
             if lookup_endpoint == endpoint:
-                old_endpoint = endpoint_owner.endpoint
-                endpoint_owner.endpoint = endpoint
-                metadata.update_host(endpoint_owner, old_endpoint)
+                self.cluster.replace_host_endpoint(
+                    endpoint_owner, endpoint,
+                    endpoint_owner.datacenter, endpoint_owner.rack)
             return get_host(lookup_endpoint)
 
         metadata.get_host = Mock(side_effect=move_owner_before_local_row)
@@ -2027,7 +2027,10 @@ def test_changed_endpoint_skip_keeps_current_host_tokens(self):
 
         def change_after_batch(moves):
             replacements = replace_hosts_endpoints(moves)
-            replacements[0].endpoint = DefaultEndPoint('192.168.1.10')
+            current = replacements[0]
+            self.cluster.replace_host_endpoint(
+                current, DefaultEndPoint('192.168.1.10'),
+                current.datacenter, current.rack)
             return replacements
 
         self.cluster.replace_hosts_endpoints = Mock(side_effect=change_after_batch)
@@ -2072,10 +2075,9 @@ def discover_after_lookup(host_id):
     def test_new_host_does_not_displace_live_unix_endpoint_owner(self):
         self.control_connection.refresh_node_list_and_token_map()
         owner = self.cluster.metadata.get_host_by_host_id(HOST_ID_2)
-        old_endpoint = owner.endpoint
         unix_endpoint = UnixSocketEndPoint('/tmp/maintenance.sock')
-        owner.endpoint = unix_endpoint
-        self.cluster.metadata.update_host(owner, old_endpoint)
+        owner = self.cluster.replace_host_endpoint(
+            owner, unix_endpoint, owner.datacenter, owner.rack)
 
         self.connection.peer_results[1].append([
             '192.168.1.9', '10.0.0.9', 'a', 'dc1', 'rack1',