diff --git a/test/topology_custom/suite.yaml b/test/topology_custom/suite.yaml
index 5da3ca23ad..76e57a8a5f 100644
--- a/test/topology_custom/suite.yaml
+++ b/test/topology_custom/suite.yaml
@@ -10,6 +10,7 @@ run_first:
   - test_raft_recovery_basic
   - test_group0_schema_versioning
   - test_tablets_migration
+  - test_zero_token_nodes_topology_ops
 skip_in_debug:
   - test_shutdown_hang
   - test_replace
@@ -19,3 +20,5 @@ run_in_dev:
   - test_group0_schema_versioning
   - test_different_group0_ids
   - test_replace_ignore_nodes
+  - test_zero_token_nodes_no_replication
+  - test_not_enough_token_owners
diff --git a/test/topology_custom/test_zero_token_nodes_topology_ops.py b/test/topology_custom/test_zero_token_nodes_topology_ops.py
new file mode 100644
index 0000000000..5dcf0dc84c
--- /dev/null
+++ b/test/topology_custom/test_zero_token_nodes_topology_ops.py
@@ -0,0 +1,110 @@
+#
+# Copyright (C) 2024-present ScyllaDB
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+import pytest
+import logging
+import time
+
+from cassandra.cluster import ConsistencyLevel
+
+from test.pylib.manager_client import ManagerClient
+from test.pylib.scylla_cluster import ReplaceConfig
+from test.pylib.util import wait_for_cql_and_get_hosts
+from test.topology.util import check_node_log_for_failed_mutations, start_writes
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('tablets_enabled', [True, False])
+async def test_zero_token_nodes_topology_ops(manager: ManagerClient, tablets_enabled: bool):
+    """
+    Test that:
+    - adding a zero-token node in the gossip-based topology fails
+    - topology operations in the Raft-based topology involving zero-token nodes succeed
+    - client requests to normal nodes in the presence of zero-token nodes (2 normal nodes, RF=2, CL=2) succeed
+    """
+    logging.info('Trying to add a zero-token server in the gossip-based topology')
+    await manager.server_add(config={'join_ring': False, 'force_gossip_topology_changes': True},
+                             expected_error='the raft-based topology is disabled')
+
+    normal_cfg = {'enable_tablets': tablets_enabled}
+    zero_token_cfg = {'enable_tablets': tablets_enabled, 'join_ring': False}
+
+    logging.info('Adding the first server')
+    server_a = await manager.server_add(config=normal_cfg)
+
+    logging.info('Adding the second server as zero-token')
+    server_b = await manager.server_add(config=zero_token_cfg)
+
+    logging.info('Adding the third server')
+    server_c = await manager.server_add(config=normal_cfg)
+
+    await wait_for_cql_and_get_hosts(manager.cql, [server_a, server_c], time.time() + 60)
+    finish_writes = await start_writes(manager.cql, 2, ConsistencyLevel.TWO)
+
+    logging.info('Adding the fourth server as zero-token')
+    await manager.server_add(config=zero_token_cfg)  # Necessary to preserve the Raft majority.
+
+    logging.info(f'Restarting {server_b}')
+    await manager.server_stop_gracefully(server_b.server_id)
+    await manager.server_start(server_b.server_id)
+
+    logging.info(f'Stopping {server_b}')
+    await manager.server_stop_gracefully(server_b.server_id)
+
+    replace_cfg_b = ReplaceConfig(replaced_id=server_b.server_id, reuse_ip_addr=False, use_host_id=False)
+    logging.info(f'Trying to replace {server_b} with a token-owing server')
+    await manager.server_add(replace_cfg_b, config=normal_cfg, expected_error='Cannot replace the zero-token node')
+
+    logging.info(f'Replacing {server_b}')
+    server_b = await manager.server_add(replace_cfg_b, config=zero_token_cfg)
+
+    logging.info(f'Stopping {server_b}')
+    await manager.server_stop_gracefully(server_b.server_id)
+
+    replace_cfg_b = ReplaceConfig(replaced_id=server_b.server_id, reuse_ip_addr=True, use_host_id=False)
+    logging.info(f'Replacing {server_b} with the same IP')
+    server_b = await manager.server_add(replace_cfg_b, config=zero_token_cfg)
+
+    logging.info(f'Decommissioning {server_b}')
+    await manager.decommission_node(server_b.server_id)
+
+    logging.info('Adding two zero-token servers')
+    [server_b, server_d] = await manager.servers_add(2, config=zero_token_cfg)
+
+    logging.info(f'Rebuilding {server_b}')
+    await manager.rebuild_node(server_b.server_id)
+
+    logging.info(f'Stopping {server_b}')
+    await manager.server_stop_gracefully(server_b.server_id)
+
+    logging.info(f'Stopping {server_d}')
+    await manager.server_stop_gracefully(server_d.server_id)
+
+    logging.info(f'Initiating removenode of {server_b} by {server_a}, ignore_dead={[server_d.ip_addr]}')
+    await manager.remove_node(server_a.server_id, server_b.server_id, [server_d.ip_addr])
+
+    logging.info(f'Initiating removenode of {server_d} by {server_a}')
+    await manager.remove_node(server_a.server_id, server_d.server_id)
+
+    logging.info('Adding a zero-token server')
+    await manager.server_add(config=zero_token_cfg)
+
+    # FIXME: Finish writes after the last server_add call once scylladb/scylladb#19737 is fixed.
+    logging.info('Checking results of the background writes')
+    await finish_writes()
+
+    logging.info('Adding a normal server')
+    server_e = await manager.server_add(config=normal_cfg)
+
+    logging.info(f'Stopping {server_e}')
+    await manager.server_stop_gracefully(server_e.server_id)
+
+    replace_cfg_e = ReplaceConfig(replaced_id=server_e.server_id, reuse_ip_addr=False, use_host_id=False)
+    logging.info(f'Trying to replace {server_e} with a zero-token server')
+    await manager.server_add(replace_cfg_e, config=zero_token_cfg,
+                             expected_error='Cannot replace the token-owning node')
+
+    await check_node_log_for_failed_mutations(manager, server_a)
+    await check_node_log_for_failed_mutations(manager, server_c)
diff --git a/test/topology_experimental_raft/test_not_enough_token_owners.py b/test/topology_experimental_raft/test_not_enough_token_owners.py
new file mode 100644
index 0000000000..b909b625f5
--- /dev/null
+++ b/test/topology_experimental_raft/test_not_enough_token_owners.py
@@ -0,0 +1,71 @@
+#
+# Copyright (C) 2024-present ScyllaDB
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+import pytest
+import logging
+import time
+
+from test.pylib.manager_client import ManagerClient
+from test.pylib.util import unique_name, wait_for_cql_and_get_hosts
+
+
+@pytest.mark.asyncio
+async def test_not_enough_token_owners(manager: ManagerClient):
+    """
+    Test that:
+    - the first node in the cluster cannot be a zero-token node
+    - removenode and decommission of the only token owner fail in the presence of zero-token nodes
+    - removenode and decommission of a token owner fail in the presence of zero-token nodes if the number of token
+      owners would fall below the RF of some keyspace using tablets
+    """
+    logging.info('Trying to add a zero-token server as the first server in the cluster')
+    await manager.server_add(config={'join_ring': False},
+                             expected_error='Cannot start the first node in the cluster as zero-token')
+
+    logging.info('Adding the first server')
+    server_a = await manager.server_add()
+
+    logging.info('Adding two zero-token servers')
+    # The second server is needed only to preserve the Raft majority.
+    server_b = (await manager.servers_add(2, config={'join_ring': False}))[0]
+
+    logging.info(f'Trying to decommission the only token owner {server_a}')
+    await manager.decommission_node(server_a.server_id,
+                                    expected_error='Cannot decommission the last token-owning node in the cluster')
+
+    logging.info(f'Stopping {server_a}')
+    await manager.server_stop_gracefully(server_a.server_id)
+
+    logging.info(f'Trying to remove the only token owner {server_a} by {server_b}')
+    await manager.remove_node(server_b.server_id, server_a.server_id,
+                              expected_error='cannot be removed because it is the last token-owning node in the cluster')
+
+    logging.info(f'Starting {server_a}')
+    await manager.server_start(server_a.server_id)
+
+    logging.info('Adding a normal server')
+    await manager.server_add()
+
+    cql = manager.get_cql()
+
+    await wait_for_cql_and_get_hosts(cql, [server_a], time.time() + 60)
+
+    ks_name = unique_name()
+    await cql.run_async(f"""CREATE KEYSPACE {ks_name} WITH replication = {{'class': 'NetworkTopologyStrategy',
+                             'replication_factor': 2}} AND tablets = {{ 'enabled': true }}""")
+    await cql.run_async(f'CREATE TABLE {ks_name}.tbl (pk int PRIMARY KEY, v int)')
+    await cql.run_async(f'INSERT INTO {ks_name}.tbl (pk, v) VALUES (1, 1)')
+
+    # FIXME: Once scylladb/scylladb#16195 is fixed, we will have to replace the expected error message.
+    # A similar change may be needed for remove_node below.
+    logging.info(f'Trying to decommission {server_a} - one of the two token owners')
+    await manager.decommission_node(server_a.server_id, expected_error='Unable to find new replica for tablet')
+
+    logging.info(f'Stopping {server_a}')
+    await manager.server_stop_gracefully(server_a.server_id)
+
+    logging.info(f'Trying to remove {server_a}, one of the two token owners, by {server_b}')
+    await manager.remove_node(server_b.server_id, server_a.server_id,
+                              expected_error='Unable to find new replica for tablet')
diff --git a/test/topology_experimental_raft/test_zero_token_nodes_multidc.py b/test/topology_experimental_raft/test_zero_token_nodes_multidc.py
new file mode 100644
index 0000000000..3c8d40f338
--- /dev/null
+++ b/test/topology_experimental_raft/test_zero_token_nodes_multidc.py
@@ -0,0 +1,87 @@
+#
+# Copyright (C) 2024-present ScyllaDB
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+import logging
+import pytest
+
+from cassandra import ConsistencyLevel
+from cassandra.policies import WhiteListRoundRobinPolicy
+from cassandra.query import SimpleStatement
+from test.pylib.manager_client import ManagerClient
+
+from test.pylib.util import unique_name
+from test.topology.conftest import cluster_con
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('zero_token_nodes', [1, 2])
+async def test_zero_token_nodes_multidc_basic(manager: ManagerClient, zero_token_nodes: int):
+    """
+    Test the basic functionality of a DC with zero-token nodes:
+    - adding zero-token nodes to a new DC succeeds
+    - with tablets, ensuring enough replicas for tables depends on the number of token-owners in a DC, not all nodes
+    - client requests in the presence of zero-token nodes succeed (also when zero-token nodes coordinate)
+    """
+    normal_cfg = {'endpoint_snitch': 'GossipingPropertyFileSnitch'}
+    zero_token_cfg = {'endpoint_snitch': 'GossipingPropertyFileSnitch', 'join_ring': False}
+    property_file_dc1 = {'dc': 'dc1', 'rack': 'rack'}
+    property_file_dc2 = {'dc': 'dc2', 'rack': 'rack'}
+
+    logging.info('Creating dc1 with 2 token-owning nodes')
+    servers = await manager.servers_add(2, config=normal_cfg, property_file=property_file_dc1)
+
+    normal_nodes_in_dc2 = 2 - zero_token_nodes
+    logging.info(f'Creating dc2 with {normal_nodes_in_dc2} token-owning and {zero_token_nodes} zero-token nodes')
+    servers += await manager.servers_add(zero_token_nodes, config=zero_token_cfg, property_file=property_file_dc2)
+    if zero_token_nodes == 1:
+        servers.append(await manager.server_add(config=normal_cfg, property_file=property_file_dc2))
+
+    logging.info('Creating connections to dc1 and dc2')
+    dc1_cql = cluster_con([servers[0].ip_addr], 9042, False,
+                          load_balancing_policy=WhiteListRoundRobinPolicy([servers[0].ip_addr])).connect()
+    dc2_cql = cluster_con([servers[2].ip_addr], 9042, False,
+                          load_balancing_policy=WhiteListRoundRobinPolicy([servers[2].ip_addr])).connect()
+
+    ks_names = list[str]()
+    logging.info('Trying to create tables for different replication factors')
+    for rf in range(3):
+        ks_names.append(unique_name())
+        failed = False
+        await dc2_cql.run_async(f"""CREATE KEYSPACE {ks_names[rf]} WITH replication =
+                                     {{'class': 'NetworkTopologyStrategy', 'replication_factor': 2, 'dc2': {rf}}}
+                                     AND tablets = {{ 'enabled': true }}""")
+        try:
+            await dc2_cql.run_async(f'CREATE TABLE {ks_names[rf]}.tbl (pk int PRIMARY KEY, v int)')
+        except Exception:
+            failed = True
+        assert failed == (rf > normal_nodes_in_dc2)
+
+    logging.info('Sending requests with different consistency levels')
+    for rf in range(normal_nodes_in_dc2 + 1):
+        # FIXME: we may add LOCAL_QUORUM to the list below once scylladb/scylladb#20028 is fixed.
+        cls = [
+            ConsistencyLevel.ONE,
+            ConsistencyLevel.TWO,
+            ConsistencyLevel.QUORUM,
+            ConsistencyLevel.EACH_QUORUM,
+            ConsistencyLevel.ALL
+        ]
+        if rf == 1:
+            cls.append(ConsistencyLevel.LOCAL_ONE)
+
+        for cl in cls:
+            insert_query = SimpleStatement(f'INSERT INTO {ks_names[rf]}.tbl (pk, v) VALUES (1, 1)',
+                                           consistency_level=cl)
+            await dc1_cql.run_async(insert_query)
+            await dc2_cql.run_async(insert_query)
+
+            if cl == ConsistencyLevel.EACH_QUORUM:
+                continue  # EACH_QUORUM is supported only for writes
+
+            select_query = SimpleStatement(f'SELECT * FROM {ks_names[rf]}.tbl', consistency_level=cl)
+            dc1_result = list((await dc1_cql.run_async(select_query))[0])
+            dc2_result = list((await dc2_cql.run_async(select_query))[0])
+            assert dc1_result == [1, 1]
+            assert dc2_result == [1, 1]
diff --git a/test/topology_experimental_raft/test_zero_token_nodes_no_replication.py b/test/topology_experimental_raft/test_zero_token_nodes_no_replication.py
new file mode 100644
index 0000000000..c183ab2e62
--- /dev/null
+++ b/test/topology_experimental_raft/test_zero_token_nodes_no_replication.py
@@ -0,0 +1,71 @@
+#
+# Copyright (C) 2024-present ScyllaDB
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+import pytest
+import logging
+
+from cassandra.cluster import ConsistencyLevel
+from cassandra.policies import WhiteListRoundRobinPolicy
+from cassandra.query import SimpleStatement
+
+from test.pylib.manager_client import ManagerClient
+from test.pylib.util import unique_name
+from test.topology.conftest import cluster_con
+
+
+@pytest.mark.asyncio
+async def test_zero_token_nodes_no_replication(manager: ManagerClient):
+    """
+    Test that zero-token nodes aren't replicas in all non-local replication strategies with and without tablets.
+    """
+    logging.info('Adding the first server')
+    server_a = await manager.server_add()
+    logging.info('Adding the second server as zero-token')
+    server_b = await manager.server_add(config={'join_ring': False})
+    logging.info('Adding the third server')
+    await manager.server_add()
+
+    logging.info(f'Initiating connections to {server_a} and {server_b}')
+    cql_a = cluster_con([server_a.ip_addr], 9042, False,
+                        load_balancing_policy=WhiteListRoundRobinPolicy([server_a.ip_addr])).connect()
+    cql_b = cluster_con([server_b.ip_addr], 9042, False,
+                        load_balancing_policy=WhiteListRoundRobinPolicy([server_b.ip_addr])).connect()
+
+    logging.info('Creating tables for each replication strategy and tablets combination')
+    ks_names = list[str]()
+    for replication_strategy in ['EverywhereStrategy', 'SimpleStrategy', 'NetworkTopologyStrategy']:
+        for tablets_enabled in [True, False]:
+            if tablets_enabled and replication_strategy != 'NetworkTopologyStrategy':
+                continue
+
+            ks_name = unique_name()
+            ks_names.append(ks_name)
+            await cql_b.run_async(f"""CREATE KEYSPACE {ks_name} WITH replication =
+                                    {{'class': '{replication_strategy}', 'replication_factor': 2}}
+                                    AND tablets = {{ 'enabled': {str(tablets_enabled).lower()} }}""")
+            await cql_b.run_async(f'CREATE TABLE {ks_name}.tbl (pk int PRIMARY KEY, v int)')
+            for i in range(100):
+                insert_query = f'INSERT INTO {ks_name}.tbl (pk, v) VALUES ({i}, {i})'
+                if i % 2 == 0:
+                    await cql_a.run_async(insert_query)
+                else:
+                    await cql_b.run_async(insert_query)  # Zero-token nodes should be able to coordinate requests.
+
+    select_queries = {ks_name: SimpleStatement(f'SELECT * FROM {ks_name}.tbl', consistency_level=ConsistencyLevel.ALL)
+                      for ks_name in ks_names}
+
+    for ks_name in ks_names:
+        result1 = [(row.pk, row.v) for row in await cql_b.run_async(select_queries[ks_name])]
+        result1.sort()
+        assert result1 == [(i, i) for i in range(100)]
+
+    logging.info(f'Stopping {server_b}')
+    await manager.server_stop_gracefully(server_b.server_id)
+
+    # If server_b was a replica of some token range or tablet, reads with CL=ALL would fail.
+    for ks_name in ks_names:
+        result2 = [(row.pk, row.v) for row in await cql_a.run_async(select_queries[ks_name])]
+        result2.sort()
+        assert result2 == [(i, i) for i in range(100)]