control-connection: keep zero-token hosts

cassandra/cluster.py

@@ -3963,9 +3963,8 @@
 
         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
 
@@ -4340,7 +4339,7 @@
                         fn, args, kwargs = task
                         kwargs = dict(kwargs)
                         future = self._executor.submit(fn, *args, **kwargs)
                         future.add_done_callback(self._log_if_failed)
                     else:
                         self._queue.put_nowait((run_at, i, task))
                         break

tests/unit/test_control_connection.py

@@ -80,8 +80,8 @@ 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
 
     def all_hosts_items(self):
         return list(self.hosts.items())
@@ -206,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
@@ -321,7 +330,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 +343,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 +418,44 @@ 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._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]
         self.control_connection.refresh_node_list_and_token_map()
@@ -589,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)