cluster: clean up failed host additions

Author: dkropachev

cassandra/cluster.py

@@ -1870,6 +1870,18 @@ def _cleanup_failed_on_up_handling(self, host):
 
         self._start_reconnector(host, is_host_addition=False)
 
+    def _cleanup_failed_on_add_handling(self, host):
+        with host.lock:
+            host.set_down()
+            host._currently_handling_node_addition = False
+
+        self.profile_manager.on_down(host)
+        self.control_connection.on_down(host)
+        for session in tuple(self.sessions):
+            session.remove_pool(host)
+
+        self._start_reconnector(host, is_host_addition=True)
+
     def _on_up_future_completed(self, host, futures, results, lock, finished_future):
         with lock:
             futures.discard(finished_future)
@@ -2100,14 +2112,12 @@ def future_completed(future):
 
                 for exc in [f for f in futures_results if isinstance(f, Exception)]:
                     log.error("Unexpected failure while adding node %s, will not mark up:", host, exc_info=exc)
-                    with host.lock:
-                        host._currently_handling_node_addition = False
+                    self._cleanup_failed_on_add_handling(host)
                     return
 
                 if not all(futures_results):
                     log.warning("Connection pool could not be created, not marking node %s up", host)
-                    with host.lock:
-                        host._currently_handling_node_addition = False
+                    self._cleanup_failed_on_add_handling(host)
                     return
 
                 self._finalize_add(host)
@@ -2127,8 +2137,7 @@ def future_completed(future):
             add_aborted = True
             for future in tuple(futures):
                 future.cancel()
-            with host.lock:
-                host._currently_handling_node_addition = False
+            self._cleanup_failed_on_add_handling(host)
             raise
 
     def _finalize_add(self, host, set_up=True):

tests/unit/test_cluster.py

@@ -103,7 +103,7 @@ def test_tuple_for_contact_points(self):
 
     def test_on_add_clears_in_progress_flag_when_later_session_add_fails(self):
         cluster = Cluster(protocol_version=4)
-        host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4())
+        host = Host("127.0.0.1", SimpleConvictionPolicy, datacenter="dc1", rack="rack1", host_id=uuid.uuid4())
         successful_session = Mock()
         successful_session.add_or_renew_pool.return_value = Future()
         successful_session.update_created_pools.return_value = set()
@@ -116,6 +116,8 @@ def test_on_add_clears_in_progress_flag_when_later_session_add_fails(self):
                 cluster.on_add(host, refresh_nodes=False)
 
             assert not host._currently_handling_node_addition
+            load_balancer = cluster.profile_manager.default.load_balancing_policy
+            assert host not in list(load_balancer.make_query_plan())
 
             with pytest.raises(RuntimeError):
                 cluster.on_add(host, refresh_nodes=False)
@@ -124,6 +126,26 @@ def test_on_add_clears_in_progress_flag_when_later_session_add_fails(self):
         finally:
             cluster.shutdown()
 
+    def test_on_add_excludes_host_from_query_plan_when_pool_future_fails(self):
+        cluster = Cluster(protocol_version=4)
+        host = Host("127.0.0.1", SimpleConvictionPolicy, datacenter="dc1", rack="rack1", host_id=uuid.uuid4())
+        failed_future = Future()
+        session = Mock()
+        session.add_or_renew_pool.return_value = failed_future
+        session.update_created_pools.return_value = set()
+        cluster.sessions = [session]
+
+        try:
+            cluster.on_add(host, refresh_nodes=False)
+
+            failed_future.set_result(False)
+
+            load_balancer = cluster.profile_manager.default.load_balancing_policy
+            assert host not in list(load_balancer.make_query_plan())
+            assert host.is_up is False
+        finally:
+            cluster.shutdown()
+
     def test_on_add_waits_for_all_session_pool_futures_before_marking_host_up(self):
         cluster = Cluster(protocol_version=4)
         host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4())