session: reject stale same-endpoint pool creation

Author: dkropachev

cassandra/cluster.py

@@ -4191,10 +4191,18 @@ def callback(pool, errors):
                                 metadata_host = host
                                 if isinstance(self.cluster.metadata, Metadata):
                                     metadata_host = self.cluster.metadata.get_host_by_host_id(host.host_id)
+                                    if metadata_host is None:
+                                        log.debug(
+                                            "Discarding stale connection pool for host %s; "
+                                            "host id is no longer present in metadata",
+                                            host)
+                                        self._invalidate_pool_creation(
+                                            host, expected_endpoint=creation_endpoint)
+                                        discard_pool = True
 
                                 target_host = metadata_host if metadata_host is not None else host
                                 target_endpoint_changed = False
-                                if target_host is not host:
+                                if not discard_pool and target_host is not host:
                                     with target_host.lock:
                                         target_endpoint_changed = not self._endpoints_match(
                                             target_host.endpoint, creation_endpoint)
@@ -4207,7 +4215,8 @@ def callback(pool, errors):
                                     self._invalidate_pool_creation(
                                         host, expected_endpoint=creation_endpoint)
                                     discard_pool = True
-                                else:
+
+                                if not discard_pool:
                                     target_host_matches = False
                                     for pool_host in tuple(retained_pools):
                                         if pool_host is target_host:

tests/unit/test_cluster.py

@@ -726,6 +726,37 @@ def make_pool(host, distance, pool_session, endpoint=None):
         assert session._pools == {}
         created_pools[0].shutdown.assert_called_once_with()
 
+    def test_stale_host_pool_creation_does_not_replace_same_endpoint_host(self):
+        endpoint = DefaultEndPoint("127.0.0.1")
+        stale_host = Host(endpoint, SimpleConvictionPolicy,
+                          host_id=uuid.uuid4())
+        replacement_host = Host(endpoint, SimpleConvictionPolicy,
+                                host_id=uuid.uuid4())
+        cluster, session, executor = self._make_cluster_and_session(
+            [replacement_host])
+        cluster.metadata = Metadata()
+        cluster.metadata.add_or_return_host(replacement_host)
+        replacement_pool = self._make_pool(
+            replacement_host, HostDistance.LOCAL, session)
+        session._pools[replacement_host] = replacement_pool
+        created_pools = []
+
+        def make_pool(host, distance, pool_session, endpoint=None):
+            pool = self._make_pool(host, distance, pool_session, endpoint)
+            created_pools.append(pool)
+            return pool
+
+        with patch("cassandra.cluster.HostConnection", side_effect=make_pool):
+            future = session.add_or_renew_pool(
+                stale_host, is_host_addition=False)
+
+            executor.run_next()
+
+        assert future.result() is False
+        assert session._pools[replacement_host] is replacement_pool
+        replacement_pool.shutdown.assert_not_called()
+        created_pools[0].shutdown.assert_called_once_with()
+
     def test_remove_pool_expected_host_mismatch_invalidates_stale_creation(self):
         stale_host = self._make_host("127.0.0.1")
         replacement_host = self._make_host("127.0.0.1")
@@ -1072,6 +1103,8 @@ def _make_session_with_pool(host, pool):
         session = Session.__new__(Session)
         session._lock = Lock()
         session._pools = {host: pool}
+        session.cluster = Mock()
+        session.cluster.metadata.all_hosts.return_value = []
         session.submit = _ImmediateExecutor().submit
         return session