From fa61d4db9c63c52720cc234e16a95bd9334ed28e Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Tue, 5 May 2026 13:15:18 -0400 Subject: [PATCH 1/2] control-connection: keep zero-token hosts --- cassandra/cluster.py | 3 +-- tests/unit/test_control_connection.py | 19 +++++++++++++++++-- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 8da9df6a55..e76ece316d 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -3963,9 +3963,8 @@ def _is_valid_peer(row): if "tokens" in row and not row.get("tokens"): log.debug( - "Found a zero-token node - tokens is None (broadcast_rpc: %s, host_id: %s). Ignoring host." % + "Found a zero-token node - tokens is None (broadcast_rpc: %s, host_id: %s). Adding host without tokens." % (broadcast_rpc, host_id)) - return False return True diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index d759e12332..631b98347f 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -82,6 +82,7 @@ def update_host(self, host, old_endpoint): host, created = self.add_or_return_host(host) self._host_id_by_endpoint[host.endpoint] = host.host_id self._host_id_by_endpoint.pop(old_endpoint, False) + self._host_id_by_endpoint[host.endpoint] = host.host_id def all_hosts_items(self): return list(self.hosts.items()) @@ -321,7 +322,6 @@ def refresh_and_validate_added_hosts(): [None, None, "a", "dc1", "rack1", ["1", "101", "201"], 'uuid1'], ["192.168.1.7", "10.0.0.1", "a", None, "rack1", ["1", "101", "201"], 'uuid2'], ["192.168.1.6", "10.0.0.1", "a", "dc1", None, ["1", "101", "201"], 'uuid3'], - ["192.168.1.5", "10.0.0.1", "a", "dc1", "rack1", None, 'uuid4'], ["192.168.1.4", "10.0.0.1", "a", "dc1", "rack1", ["1", "101", "201"], None]]]) refresh_and_validate_added_hosts() @@ -335,7 +335,6 @@ def refresh_and_validate_added_hosts(): [None, 9042, None, 7040, "a", "dc1", "rack1", ["2", "102", "202"], "uuid2"], ["192.168.1.5", 9042, "10.0.0.2", 7040, "a", None, "rack1", ["2", "102", "202"], "uuid2"], ["192.168.1.5", 9042, "10.0.0.2", 7040, "a", "dc1", None, ["2", "102", "202"], "uuid2"], - ["192.168.1.5", 9042, "10.0.0.2", 7040, "a", "dc1", "rack1", None, "uuid2"], ["192.168.1.5", 9042, "10.0.0.2", 7040, "a", "dc1", "rack1", ["2", "102", "202"], None]]]) refresh_and_validate_added_hosts() @@ -411,6 +410,22 @@ def test_refresh_nodes_and_tokens_add_host(self): assert self.cluster.added_hosts[0].rack == "rack1" assert self.cluster.added_hosts[0].host_id == "uuid4" + def test_refresh_nodes_and_tokens_adds_zero_token_host_without_token_map_entry(self): + # Zero-token nodes are valid topology members, but they do not own token ranges. + self.connection.peer_results[1].append( + ["192.168.1.3", "10.0.0.3", "a", "dc1", "rack1", None, "uuid4"] + ) + self.cluster.scheduler.schedule = lambda delay, f, *args, **kwargs: f(*args, **kwargs) + + self.control_connection.refresh_node_list_and_token_map() + + zero_token_host = self.cluster.metadata.get_host(DefaultEndPoint("192.168.1.3")) + assert zero_token_host is not None + assert zero_token_host.host_id == "uuid4" + assert zero_token_host.datacenter == "dc1" + assert zero_token_host.rack == "rack1" + assert zero_token_host not in self.cluster.metadata.token_map + def test_refresh_nodes_and_tokens_remove_host(self): del self.connection.peer_results[1][1] self.control_connection.refresh_node_list_and_token_map() From 5d225d68203a741ace9e37adfea3ae8aa5d4e300 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Tue, 5 May 2026 13:53:46 -0400 Subject: [PATCH 2/2] test: expand zero-token host coverage --- tests/unit/test_control_connection.py | 67 ++++++++++++++++++++++++--- 1 file changed, 60 insertions(+), 7 deletions(-) diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index 631b98347f..b8be438bf4 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -80,7 +80,6 @@ def add_or_return_host(self, host): def update_host(self, host, old_endpoint): host, created = self.add_or_return_host(host) - self._host_id_by_endpoint[host.endpoint] = host.host_id self._host_id_by_endpoint.pop(old_endpoint, False) self._host_id_by_endpoint[host.endpoint] = host.host_id @@ -207,6 +206,15 @@ def setUp(self): self.control_connection._connection = self.connection self.control_connection._time = self.time + def _assert_zero_token_host_without_token_map_entry(self, endpoint, host_id): + zero_token_host = self.cluster.metadata.get_host(endpoint) + assert zero_token_host is not None + assert zero_token_host.host_id == host_id + assert zero_token_host.datacenter == "dc1" + assert zero_token_host.rack == "rack1" + assert zero_token_host not in self.cluster.metadata.token_map + return zero_token_host + def test_wait_for_schema_agreement(self): """ Basic test with all schema versions agreeing @@ -419,12 +427,34 @@ def test_refresh_nodes_and_tokens_adds_zero_token_host_without_token_map_entry(s self.control_connection.refresh_node_list_and_token_map() - zero_token_host = self.cluster.metadata.get_host(DefaultEndPoint("192.168.1.3")) - assert zero_token_host is not None - assert zero_token_host.host_id == "uuid4" - assert zero_token_host.datacenter == "dc1" - assert zero_token_host.rack == "rack1" - assert zero_token_host not in self.cluster.metadata.token_map + zero_token_host = self._assert_zero_token_host_without_token_map_entry( + DefaultEndPoint("192.168.1.3"), "uuid4") + assert 1 == len(self.cluster.added_hosts) + assert self.cluster.added_hosts[0] is zero_token_host + assert [] == self.cluster.metadata.removed_hosts + + def test_refresh_nodes_and_tokens_adds_empty_token_host_without_token_map_entry(self): + self.connection.peer_results[1].append( + ["192.168.1.3", "10.0.0.3", "a", "dc1", "rack1", [], "uuid4"] + ) + self.cluster.scheduler.schedule = lambda delay, f, *args, **kwargs: f(*args, **kwargs) + + self.control_connection.refresh_node_list_and_token_map() + + zero_token_host = self._assert_zero_token_host_without_token_map_entry( + DefaultEndPoint("192.168.1.3"), "uuid4") + assert 1 == len(self.cluster.added_hosts) + assert self.cluster.added_hosts[0] is zero_token_host + + def test_refresh_nodes_and_tokens_keeps_zero_token_local_host_without_token_map_entry(self): + self.connection.local_results[1][0][7] = None + + self.control_connection.refresh_node_list_and_token_map() + + self._assert_zero_token_host_without_token_map_entry( + DefaultEndPoint("192.168.1.0"), "uuid1") + assert [] == self.cluster.added_hosts + assert [] == self.cluster.metadata.removed_hosts def test_refresh_nodes_and_tokens_remove_host(self): del self.connection.peer_results[1][1] @@ -604,6 +634,29 @@ def test_refresh_nodes_and_tokens_add_host_detects_port(self): assert self.cluster.added_hosts[0].datacenter == "dc1" assert self.cluster.added_hosts[0].rack == "rack1" + def test_refresh_nodes_and_tokens_adds_zero_token_host_from_peers_v2_without_token_map_entry(self): + del self.connection.peer_results[:] + self.connection.peer_results.extend(self.connection.peer_results_v2) + self.connection.peer_results[1].append( + ["192.168.1.3", 555, "10.0.0.3", 666, "a", "dc1", "rack1", None, "uuid4"] + ) + self.connection.wait_for_responses = Mock(return_value=_node_meta_results( + self.connection.local_results, self.connection.peer_results)) + self.cluster.scheduler.schedule = lambda delay, f, *args, **kwargs: f(*args, **kwargs) + + self.control_connection.refresh_node_list_and_token_map() + + zero_token_host = self._assert_zero_token_host_without_token_map_entry( + DefaultEndPoint("192.168.1.3", 555), "uuid4") + assert 1 == len(self.cluster.added_hosts) + assert self.cluster.added_hosts[0] is zero_token_host + assert zero_token_host.endpoint.port == 555 + assert zero_token_host.broadcast_rpc_address == "192.168.1.3" + assert zero_token_host.broadcast_rpc_port == 555 + assert zero_token_host.broadcast_address == "10.0.0.3" + assert zero_token_host.broadcast_port == 666 + assert [] == self.cluster.metadata.removed_hosts + def test_refresh_nodes_and_tokens_add_host_detects_invalid_port(self): del self.connection.peer_results[:] self.connection.peer_results.extend(self.connection.peer_results_v2)