cluster: fence up pool creation by endpoint

Author: dkropachev

cassandra/cluster.py

@@ -2455,7 +2455,9 @@ def _clear_stale_reconnector():
                                futures, futures_results, futures_lock)
             for session in tuple(self.sessions):
                 future = session.add_or_renew_pool(
-                    host, is_host_addition=False, allow_retry_after_auth_failure=True)
+                    host, is_host_addition=False,
+                    allow_retry_after_auth_failure=True,
+                    expected_endpoint=up_handling_endpoint)
                 if future is not None:
                     have_future = True
                     futures.add(future)
@@ -4076,7 +4078,9 @@ def _invalidate_pool_creation(self, host, expected_endpoint=None):
             return True
         return False
 
-    def add_or_renew_pool(self, host, is_host_addition, allow_retry_after_auth_failure=False):
+    def add_or_renew_pool(self, host, is_host_addition,
+                          allow_retry_after_auth_failure=False,
+                          expected_endpoint=None):
         """
         For internal use only.
         """
@@ -4248,20 +4252,24 @@ def callback(pool, errors):
             return True
 
         with self._lock:
+            with host.lock:
+                creation_endpoint = host.endpoint
+                if (expected_endpoint is not None and
+                        not self._endpoints_match(
+                            creation_endpoint, expected_endpoint)):
+                    return None
+
             state = self._get_pool_creation_state(host)
             if state.creation_epoch is not None:
-                with host.lock:
-                    endpoint_changed = not self._endpoints_match(
-                        host.endpoint, state.endpoint)
+                endpoint_changed = not self._endpoints_match(
+                    creation_endpoint, state.endpoint)
                 if not endpoint_changed:
                     return state.future
                 self._invalidate_pool_creation(
                     host, expected_endpoint=state.endpoint)
 
             creation_epoch = state.advance()
             state.creation_epoch = creation_epoch
-            with host.lock:
-                creation_endpoint = host.endpoint
             state.endpoint = creation_endpoint
             future = self.submit(run_add_or_renew_pool)
             if future is None:

tests/unit/test_cluster.py

@@ -579,6 +579,40 @@ def make_pool(host, distance, pool_session, endpoint=None):
         created_pools[1].shutdown.assert_not_called()
         assert session._pools[host] is created_pools[1]
 
+    def test_on_up_does_not_publish_replacement_endpoint_pool_after_endpoint_swap(self):
+        host = self._make_host("127.0.0.1")
+        host.set_down()
+        old_endpoint = host.endpoint
+        new_endpoint = DefaultEndPoint("127.0.0.2")
+        cluster, session, executor = self._make_cluster_and_session([host])
+        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
+
+        original_add_or_renew_pool = Session.add_or_renew_pool.__get__(
+            session, Session)
+
+        def add_after_endpoint_swap(host, *args, **kwargs):
+            host.endpoint = new_endpoint
+            return original_add_or_renew_pool(host, *args, **kwargs)
+
+        session.add_or_renew_pool = Mock(side_effect=add_after_endpoint_swap)
+
+        with patch("cassandra.cluster.HostConnection", side_effect=make_pool):
+            Cluster.on_up(cluster, host)
+            while executor.submissions:
+                executor.run_next()
+
+        assert created_pools == []
+        assert session._pools == {}
+        session.add_or_renew_pool.assert_called_once_with(
+            host, is_host_addition=False,
+            allow_retry_after_auth_failure=True,
+            expected_endpoint=old_endpoint)
+
     def test_pool_creation_publishes_before_endpoint_lock_is_released(self):
         host = self._make_host("127.0.0.1")
         new_endpoint = DefaultEndPoint("127.0.0.2")
@@ -1750,7 +1784,8 @@ def queue_up_then_fail(h):
         cluster.control_connection.on_up.assert_called_once_with(host)
         session.add_or_renew_pool.assert_called_once_with(
             host, is_host_addition=False,
-            allow_retry_after_auth_failure=True)
+            allow_retry_after_auth_failure=True,
+            expected_endpoint=host.endpoint)
         assert host.is_up
         assert self._state(cluster, host).up_epoch is None
         assert self._state(cluster, host).pending_up_epoch is None