[PYTHON-1181] Stop reconnection attempts after node removal

Author: TheRealFalcon

CHANGELOG.rst

@@ -19,6 +19,7 @@ Bug Fixes
 ---------
 * re-raising the CQLEngineException will fail on Python 3 (PYTHON-1166)
 * asyncio message chunks can be processed discontinuously (PYTHON-1185)
+* Reconnect attempts persist after downed node removed from peers (PYTHON-1181)
 
 Others
 ------

cassandra/cluster.py

@@ -2029,6 +2029,10 @@ def on_remove(self, host):
             listener.on_remove(host)
         self.control_connection.on_remove(host)
 
+        reconnection_handler = host.get_and_set_reconnection_handler(None)
+        if reconnection_handler:
+            reconnection_handler.cancel()
+
     def signal_connection_failure(self, host, connection_exc, is_host_addition, expect_host_to_be_down=False):
         is_down = host.signal_connection_failure(connection_exc)
         if is_down:

tests/integration/long/test_topology_change.py

@@ -0,0 +1,49 @@
+from unittest import TestCase
+
+from cassandra.cluster import Cluster
+from cassandra.policies import HostStateListener
+from tests.integration import PROTOCOL_VERSION, get_node, use_cluster, local
+from tests.integration.long.utils import decommission
+from tests.integration.util import wait_until
+
+
+class StateListener(HostStateListener):
+    def __init__(self):
+        self.downed_host = None
+        self.removed_host = None
+
+    def on_remove(self, host):
+        self.removed_host = host
+
+    def on_up(self, host):
+        pass
+
+    def on_down(self, host):
+        self.downed_host = host
+
+    def on_add(self, host):
+        pass
+
+
+class TopologyChangeTests(TestCase):
+    @local
+    def test_removed_node_stops_reconnecting(self):
+        """ Ensure we stop reconnecting after a node is removed. PYTHON-1181 """
+        use_cluster("test_down_then_removed", [3], start=True)
+
+        state_listener = StateListener()
+        cluster = Cluster(protocol_version=PROTOCOL_VERSION)
+        self.addCleanup(cluster.shutdown)
+        cluster.register_listener(state_listener)
+        session = cluster.connect(wait_for_all_pools=True)
+
+        get_node(3).nodetool("disablebinary")
+
+        wait_until(condition=lambda: state_listener.downed_host is not None, delay=2, max_attempts=50)
+        self.assertTrue(state_listener.downed_host.is_currently_reconnecting())
+
+        decommission(3)
+
+        wait_until(condition=lambda: state_listener.removed_host is not None, delay=2, max_attempts=50)
+        self.assertIs(state_listener.downed_host, state_listener.removed_host)  # Just a sanity check
+        self.assertFalse(state_listener.removed_host.is_currently_reconnecting())