# ./test.py --mode dev test/cluster/test_maintenance_mode.py
Test session starts (platform: linux, Python 3.14.6, pytest 8.4.2, pytest-sugar 1.1.1)
rootdir: /home/avi/scylla/test
configfile: pytest.ini
plugins: allure-pytest-2.16.0, xdist-3.8.0, asyncio-1.1.0, sugar-1.1.1, timeout-2.4.0
asyncio: mode=Mode.AUTO, asyncio_default_fixture_loop_scope=session, asyncio_default_test_loop_scope=function
timeout: 3600.0s
timeout method: signal
timeout func_only: False
session timeout: 24000.0s
31 workers [1 item] collecting ...
...
―――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――― test_maintenance_mode.dev.1 ――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――
[gw30] linux -- Python 3.14.6 /usr/sbin/python3
self = <ClientResponse(http://api/cluster/server/1/start) [None None]>
None
connection = Connection<ConnectionKey(host='api', port=80, is_ssl=False, ssl=True, proxy=None, proxy_auth=None, proxy_headers_hash=None)>
async def start(self, connection: "Connection") -> "ClientResponse":
"""Start response processing."""
self._closed = False
self._protocol = connection.protocol
self._connection = connection
with self._timer:
while True:
# read response
try:
protocol = self._protocol
> message, payload = await protocol.read() # type: ignore[union-attr]
^^^^^^^^^^^^^^^^^^^^^
/usr/lib64/python3.14/site-packages/aiohttp/client_reqrep.py:539:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <aiohttp.client_proto.ResponseHandler object at 0x7fa7617e0d40>
async def read(self) -> _T:
if not self._buffer and not self._eof:
assert not self._waiter
self._waiter = self._loop.create_future()
try:
> await self._waiter
E asyncio.exceptions.CancelledError
/usr/lib64/python3.14/site-packages/aiohttp/streams.py:707: CancelledError
The above exception was the direct cause of the following exception:
manager = <test.pylib.manager_client.ManagerClient object at 0x7fa76107a3c0>
async def test_maintenance_mode(manager: ManagerClient):
"""
The test checks that in maintenance mode server A is not available for other nodes and for clients.
It is possible to connect by the maintenance socket to server A and perform local CQL operations.
The test is run with multiple keyspaces with different configurations (replication strategy, RF, tablets enabled).
It initially used only SimpleStrategy and RF=1, which hid https://github.com/scylladb/scylladb/issues/27988. To keep
the test fast, the tasks for different keyspaces are performed concurrently, and server A is started in maintenance
mode only once.
"""
max_rf = 3
servers = await manager.servers_add(max_rf, auto_rack_dc='dc1')
server_a = servers[0]
host_id_a = await manager.get_host_id(server_a.server_id)
socket_endpoint = UnixSocketEndPoint(await manager.server_get_maintenance_socket_path(server_a.server_id))
# For the move_tablet API.
await manager.disable_tablet_balancing()
# An exclusive connection to server A is needed for requests with LocalStrategy.
cluster = cluster_con([server_a.ip_addr], load_balancing_policy=WhiteListRoundRobinPolicy([server_a.ip_addr]))
cql = cluster.connect()
# (replication strategy, Optional[replication factor], tablets enabled)
KeyspaceOptions: TypeAlias = tuple[str, int | None, bool]
keyspace_options: list[KeyspaceOptions] = []
keyspace_options.append(('EverywhereStrategy', None, False))
keyspace_options.append(('LocalStrategy', None, False))
for rf in range(1, max_rf + 1):
keyspace_options.append(('SimpleStrategy', rf, False))
for tablets_enabled in [True, False]:
keyspace_options.append(('NetworkTopologyStrategy', rf, tablets_enabled))
key_on_server_a_per_table: dict[str, int] = dict()
async def prepare_table(options: KeyspaceOptions):
replication_strategy, rf, tablets_enabled = options
rf_string = "" if rf is None else f", 'replication_factor': {rf}"
ks = await create_new_test_keyspace(cql,
f"""WITH REPLICATION = {{'class': '{replication_strategy}'{rf_string}}}
AND tablets = {{'enabled': {str(tablets_enabled).lower()}, 'initial': 1}}""")
rf_tag = "" if rf is None else f"rf{rf}"
tablets_tag = "tablets" if tablets_enabled else "vnodes"
table_suffix = f"{replication_strategy.lower()}_{rf_tag}_{tablets_tag}"
table = f"{ks}.{table_suffix}"
await cql.run_async(f"CREATE TABLE {table} (k int PRIMARY KEY, v int)")
logger.info(f"Created table {table}")
async def insert_one(cl: ConsistencyLevel):
key = 1
insert_stmt = SimpleStatement(f"INSERT INTO {table} (k, v) VALUES ({key}, {key})",
consistency_level=cl)
await cql.run_async(insert_stmt)
key_on_server_a_per_table[table] = key
if replication_strategy == 'LocalStrategy':
await insert_one(ConsistencyLevel.ONE)
return
if tablets_enabled:
await insert_one(ConsistencyLevel.ALL)
logger.info(f"Ensuring that a tablet replica is on {server_a} for table {table}")
[tablet] = await get_all_tablet_replicas(manager, server_a, ks, table_suffix)
if host_id_a not in [r[0] for r in tablet.replicas]:
assert rf < max_rf
any_replica = tablet.replicas[0]
logger.info(f"Moving tablet from {any_replica} to {server_a} for table {table}")
await manager.api.move_tablet(server_a.ip_addr, ks, table_suffix,
any_replica[0], any_replica[1],
host_id_a, 0,
tablet.last_token)
return
# This path is executed only for vnodes-based keyspaces.
# Token ranges of the server A
# [(start_token, end_token)]
ranges = [(int(row[0]), int(row[1])) for row in await cql.run_async(f"""SELECT start_token, end_token
FROM system.token_ring WHERE keyspace_name = '{ks}'
AND endpoint = '{server_a.ip_addr}' ALLOW FILTERING""")]
# Insert data to the cluster until a key is stored on server A.
new_key = 0
while table not in key_on_server_a_per_table:
if new_key == 1000:
# The probability of reaching this code is (2/3)^1000 for RF=1 and lower for greater RFs. This is much
# less than, for example, the probability of a UUID collision, so worrying about this would be silly.
# It could still happen due to a bug, and then we want to know about it, so we fail the test.
pytest.fail(f"Could not find a key on server {server_a} after inserting 1000 keys")
new_key += 1
insert_stmt = SimpleStatement(f"INSERT INTO {table} (k, v) VALUES ({new_key}, {new_key})",
consistency_level=ConsistencyLevel.ALL)
await cql.run_async(insert_stmt)
res = await cql.run_async(f"SELECT token(k) FROM {table} WHERE k = {new_key}")
assert len(res) == 1
token = res[0][0]
for start, end in ranges:
if (start < end and start < token <= end) or (start >= end and (token <= end or start < token)):
logger.info(f"Found key {new_key} with token {token} on server {server_a} for table {table}")
key_on_server_a_per_table[table] = new_key
logger.info("Preparing tables")
await gather_safely(*(prepare_table(options) for options in keyspace_options))
# Start server A in maintenance mode
await manager.server_stop_gracefully(server_a.server_id)
await manager.server_update_config(server_a.server_id, "maintenance_mode", True)
> await manager.server_start(server_a.server_id)
test/cluster/test_maintenance_mode.py:137:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
test/pylib/manager_client.py:474: in server_start
await self.client.put_json(f"/cluster/server/{server_id}/start", data, timeout=timeout)
test/pylib/rest_client.py:122: in put_json
ret = await self._fetch("PUT", resource_uri, response_type = response_type, host = host,
test/pylib/rest_client.py:73: in _fetch
async with request(method, uri,
/usr/lib64/python3.14/site-packages/aiohttp/client.py:1552: in __aenter__
self._resp = await self._coro
^^^^^^^^^^^^^^^^
/usr/lib64/python3.14/site-packages/aiohttp/client.py:788: in _request
resp = await handler(req)
^^^^^^^^^^^^^^^^^^
/usr/lib64/python3.14/site-packages/aiohttp/client.py:766: in _connect_and_send_request
await resp.start(conn)
/usr/lib64/python3.14/site-packages/aiohttp/client_reqrep.py:534: in start
with self._timer:
^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <aiohttp.helpers.TimerContext object at 0x7fa76111fb40>, exc_type = <class 'asyncio.exceptions.CancelledError'>, exc_val = CancelledError(), exc_tb = <traceback object at 0x7fa75acbc940>
def __exit__(
self,
exc_type: Optional[Type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType],
) -> Optional[bool]:
enter_task: Optional[asyncio.Task[Any]] = None
if self._tasks:
enter_task = self._tasks.pop()
if exc_type is asyncio.CancelledError and self._cancelled:
assert enter_task is not None
# The timeout was hit, and the task was cancelled
# so we need to uncancel the last task that entered the context manager
# since the cancellation should not leak out of the context manager
if sys.version_info >= (3, 11):
# If the task was already cancelling don't raise
# asyncio.TimeoutError and instead return None
# to allow the cancellation to propagate
if enter_task.uncancel() > self._cancelling:
return None
> raise asyncio.TimeoutError from exc_val
E TimeoutError
/usr/lib64/python3.14/site-packages/aiohttp/helpers.py:713: TimeoutError
...
Results (436.83s (0:07:16)):
1 failed
- cluster/test_maintenance_mode.py:27 test_maintenance_mode.dev.1
Driver name ScyllaDB Python Driver, version 3.29.11
CPU utilization: 1.1%Reverting to 3.29.7 fixed the problem.
Reverting to 3.29.7 fixed the problem.