Antalya-26.3 - Backport flaky-fix commits from upstream (2026-08-07)

tests/config/config.d/storage_conf.xml

@@ -112,7 +112,13 @@
             <local_remote>
                 <volumes>
                     <local><disk>default</disk></local>
-                    <remote><disk>s3_disk</disk></remote>
+                    <!-- A part always starts on the 'local' volume, even when its move TTL
+                         expired while the INSERT was still running. The background mover
+                         ignores this setting, so TTL moves still happen. -->
+                    <remote>
+                        <disk>s3_disk</disk>
+                        <perform_ttl_move_on_insert>0</perform_ttl_move_on_insert>
+                    </remote>
                 </volumes>
             </local_remote>
             <s3_cache>

tests/integration/test_dirty_pages_force_purge/configs/overrides.yaml

@@ -1,3 +1,7 @@
 ---
 max_server_memory_usage: 4Gi
-memory_worker_purge_dirty_pages_threshold_ratio: 0.2
+# The threshold must be reliably exceeded by a single iteration of the test query
+# (peak usage ~417 MiB). With a higher ratio, `pdirty` may plateau below the threshold,
+# because jemalloc reuses dirty pages within the same arenas across iterations,
+# and the number of touched arenas depends on the machine and the number of threads.
+memory_worker_purge_dirty_pages_threshold_ratio: 0.05

tests/integration/test_hedged_requests/test.py

@@ -7,6 +7,7 @@
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from helpers.cluster import ClickHouseCluster
+from helpers.network import PartitionManager
 from helpers.test_tools import TSV
 
 cluster = ClickHouseCluster(__file__)
@@ -429,28 +430,46 @@ def test_async_connect(started_cluster):
         Distributed('test_cluster_connect', 'default', 'test_hedged')"""
     )
 
-    NODES["node"].query(
-        "SELECT hostName(), id FROM distributed_connect ORDER BY id LIMIT 1 SETTINGS prefer_localhost_replica = 0, connect_timeout_with_failover_ms=5000, async_query_sending_for_remote=0, max_threads=1, max_distributed_connections=1"
-    )
-    check_changing_replica_events(2)
-    check_if_query_sending_was_not_suspended()
-
-    # Restart server to reset connection pool state
-    NODES["node"].restart_clickhouse()
+    # The first replica of each shard in test_cluster_connect is an unreachable
+    # address (129.0.0.1 / 129.0.0.2). Silently drop the initiator's packets to
+    # them so the connect always stalls and is preempted by the
+    # hedged_connection_timeout_ms timer (the path HedgedRequestsChangeReplica
+    # counts). Otherwise, on slow builds the connect can fail fast (host
+    # unreachable), switching the replica through the connection-failure path,
+    # which does not increment that event.
+    with PartitionManager() as pm:
+        for unreachable_ip in ("129.0.0.1", "129.0.0.2"):
+            pm.add_rule(
+                {
+                    "instance": NODES["node"],
+                    "chain": "OUTPUT",
+                    "destination": unreachable_ip,
+                    "action": "DROP",
+                }
+            )
 
-    attempt = 0
-    while attempt < 100:
         NODES["node"].query(
-            "SELECT hostName(), id FROM distributed_connect ORDER BY id LIMIT 1 SETTINGS prefer_localhost_replica = 0, connect_timeout_with_failover_ms=5000, async_query_sending_for_remote=1, max_threads=1, max_distributed_connections=1"
+            "SELECT hostName(), id FROM distributed_connect ORDER BY id LIMIT 1 SETTINGS prefer_localhost_replica = 0, connect_timeout_with_failover_ms=5000, async_query_sending_for_remote=0, max_threads=1, max_distributed_connections=1"
         )
-
         check_changing_replica_events(2)
-        if check_if_query_sending_was_suspended():
-            break
+        check_if_query_sending_was_not_suspended()
 
-        attempt += 1
+        # Restart server to reset connection pool state
+        NODES["node"].restart_clickhouse()
 
-    assert attempt < 100
+        attempt = 0
+        while attempt < 100:
+            NODES["node"].query(
+                "SELECT hostName(), id FROM distributed_connect ORDER BY id LIMIT 1 SETTINGS prefer_localhost_replica = 0, connect_timeout_with_failover_ms=5000, async_query_sending_for_remote=1, max_threads=1, max_distributed_connections=1"
+            )
+
+            check_changing_replica_events(2)
+            if check_if_query_sending_was_suspended():
+                break
+
+            attempt += 1
+
+        assert attempt < 100
 
     NODES["node"].query("DROP TABLE distributed_connect")
 

tests/integration/test_keeper_auth/test.py

@@ -61,6 +61,22 @@ def zk_stop_and_close(zk):
         zk.close()
 
 
+# ZooKeeper checks delete against the PARENT's ACL, and its setACL leaves the
+# outstanding change record carrying the old ACL (duplicate() copies acl verbatim
+# and updates only stat.aversion), so a delete issued right after a widening
+# setACL can still be authorized against the pre-setACL ACL and get NoAuthError.
+def zk_delete_after_acl_change(zk, path, timeout=30.0):
+    deadline = time.monotonic() + timeout
+    while True:
+        try:
+            zk.delete(path)
+            return
+        except NoAuthError:
+            if time.monotonic() >= deadline:
+                raise
+            time.sleep(0.1)
+
+
 @pytest.mark.parametrize(("get_zk"), [get_genuine_zk, get_fake_zk])
 def test_remove_acl(started_cluster, get_zk):
     auth_connection = None
@@ -412,7 +428,9 @@ def test_partial_auth(started_cluster, get_zk):
         )
         auth_connection.set_acls("/test_partial_acl_delete", acls=[acl])
         auth_connection.set_acls("/test_partial_acl_delete/subnode", acls=[acl])
-        auth_connection.delete("/test_partial_acl_delete/subnode")
+        zk_delete_after_acl_change(auth_connection, "/test_partial_acl_delete/subnode")
+        # Authorized against "/", whose ACL this test never touches, so a
+        # NoAuthError here would be a real problem: do not retry it.
         auth_connection.delete("/test_partial_acl_delete")
         zk_stop_and_close(auth_connection)
 

tests/integration/test_keeper_back_to_back/test.py

@@ -1,6 +1,7 @@
 import os
 import random
 import string
+import threading
 import time
 from multiprocessing.dummy import Pool
 
@@ -755,9 +756,19 @@ def test_concurrent_watches(started_cluster, request):
         all_paths_triggered = []
 
         existing_path = []
+        # A watch is one-shot per (path, session), so a mutation only notifies a registration
+        # that is live at that moment. Mutating a path this thread does not hold a token for can
+        # therefore consume another thread's registration, leaving it forever unnotified.
+        existing_path_lock = threading.Lock()
         all_paths_created = []
         watches_created = 0
 
+        def claim_path():
+            with existing_path_lock:
+                if not existing_path:
+                    return None
+                return existing_path.pop(random.randrange(len(existing_path)))
+
         def create_path_and_watch(i):
             nonlocal watches_created
             nonlocal all_paths_created
@@ -775,34 +786,28 @@ def dumb_watch(event):
             fake_zk.get(global_path + "/" + str(i), watch=dumb_watch)
             all_paths_created.append(global_path + "/" + str(i))
             watches_created += 1
-            existing_path.append(i)
+            with existing_path_lock:
+                existing_path.append(i)
 
         trigger_called = 0
 
         def trigger_watch(i):
             nonlocal trigger_called
             trigger_called += 1
             fake_zk.set(global_path + "/" + str(i), b"somevalue")
-            try:
-                existing_path.remove(i)
-            except:
-                pass
 
         def call(total):
             for i in range(total):
                 create_path_and_watch(random.randint(0, 1000))
                 time.sleep(random.random() % 0.5)
-                try:
-                    rand_num = random.choice(existing_path)
-                    trigger_watch(rand_num)
-                except:
-                    pass
-            while existing_path:
-                try:
-                    rand_num = random.choice(existing_path)
+                rand_num = claim_path()
+                if rand_num is not None:
                     trigger_watch(rand_num)
-                except:
-                    pass
+            while True:
+                rand_num = claim_path()
+                if rand_num is None:
+                    break
+                trigger_watch(rand_num)
 
         p = Pool(10)
         arguments = [100] * 10

tests/integration/test_keeper_dynamic_log_level/test.py

@@ -61,21 +61,24 @@ def test_adjust_log_level(start_cluster):
             """,
         ]
     )
-    time.sleep(3)
-    node.query(
-        "SELECT * FROM system.zookeeper SETTINGS allow_unrestricted_reads_from_keeper = 'true'"
-    )
-    node.exec_in_container(
-        [
-            "bash",
-            "-c",
-            "sync",
-        ],
-        privileged=True,
-        user="root",
-    )
-    assert (
-        int(
+    # The config reloader applies the new logger settings asynchronously (it polls the config
+    # every couple of seconds), so poll until trace logging becomes active instead of relying
+    # on a fixed sleep, which is not enough on slow (e.g. sanitizer) runs.
+    trace_lines = 0
+    for _ in range(60):
+        node.query(
+            "SELECT * FROM system.zookeeper SETTINGS allow_unrestricted_reads_from_keeper = 'true'"
+        )
+        node.exec_in_container(
+            [
+                "bash",
+                "-c",
+                "sync",
+            ],
+            privileged=True,
+            user="root",
+        )
+        trace_lines = int(
             node.exec_in_container(
                 [
                     "bash",
@@ -86,5 +89,7 @@ def test_adjust_log_level(start_cluster):
                 user="root",
             )
         )
-        >= 1
-    )
+        if trace_lines >= 1:
+            break
+        time.sleep(1)
+    assert trace_lines >= 1

tests/integration/test_named_collections_encrypted2/test.py

@@ -62,6 +62,7 @@ def wait_not_exists(node, collection, timeout=10):
 
 
 def check_encrypted(zk, collection):
+    zk.sync(ZK_PATH)
     content = zk.get(f"{ZK_PATH}/{collection}.sql")[0]
     assert content[:3] == b"ENC"
     return content
@@ -717,6 +718,7 @@ def test_new_replica_encrypted_data_integrity(stopped_node3):
         password='P@ssw0rd!Complex#123'
     """)
 
+    zk.sync(ZK_PATH)
     content = zk.get(f"{ZK_PATH}/encrypted_coll.sql")[0]
     assert content[:3] == b"ENC"
     assert b"super_secret_api_key_12345" not in content

tests/integration/test_replication_credentials/test.py

@@ -46,13 +46,13 @@ def same_credentials_cluster():
 
 def test_same_credentials(same_credentials_cluster):
     node1.query("insert into test_table values ('2017-06-16', 111, 0)")
-    time.sleep(1)
+    node2.query("SYSTEM SYNC REPLICA test_table", timeout=60)
 
     assert node1.query("SELECT id FROM test_table order by id") == "111\n"
     assert node2.query("SELECT id FROM test_table order by id") == "111\n"
 
     node2.query("insert into test_table values ('2017-06-17', 222, 1)")
-    time.sleep(1)
+    node1.query("SYSTEM SYNC REPLICA test_table", timeout=60)
 
     assert node1.query("SELECT id FROM test_table order by id") == "111\n222\n"
     assert node2.query("SELECT id FROM test_table order by id") == "111\n222\n"
@@ -85,13 +85,13 @@ def no_credentials_cluster():
 
 def test_no_credentials(no_credentials_cluster):
     node3.query("insert into test_table values ('2017-06-18', 111, 0)")
-    time.sleep(1)
+    node4.query("SYSTEM SYNC REPLICA test_table", timeout=60)
 
     assert node3.query("SELECT id FROM test_table order by id") == "111\n"
     assert node4.query("SELECT id FROM test_table order by id") == "111\n"
 
     node4.query("insert into test_table values ('2017-06-19', 222, 1)")
-    time.sleep(1)
+    node3.query("SYSTEM SYNC REPLICA test_table", timeout=60)
 
     assert node3.query("SELECT id FROM test_table order by id") == "111\n222\n"
     assert node4.query("SELECT id FROM test_table order by id") == "111\n222\n"

tests/integration/test_s3_table_functions/test.py

@@ -78,14 +78,25 @@ def test_s3_table_functions(started_cluster):
 
 def test_s3_table_functions_timeouts(started_cluster):
     """
-    Test with timeout limit of 1200ms.
-    This should raise an Exception and pass.
+    A 1200ms network delay must make the S3 write time out and raise.
     """
 
+    # Make the S3 request timeout (not the connect timeout) the single failure mechanism:
+    # disable adaptive timeouts and keep the connect timeout above the delay, so the write
+    # can only fail via s3_request_timeout_ms. This exercises the send/receive idleness
+    # timeout that applies to every attempt on both fresh and reused (pooled keep-alive)
+    # connections, which is the path that was silently not timing out before.
+    timeout_settings = {
+        **settings,
+        "s3_use_adaptive_timeouts": "0",
+        "s3_connect_timeout_ms": "10000",
+        "s3_request_timeout_ms": "500",
+    }
+
     with PartitionManager() as pm:
         pm.add_network_delay(node, 1200)
 
-        with pytest.raises(QueryRuntimeException):
+        with pytest.raises(QueryRuntimeException, match="Timeout"):
             node.query(
                 """
                 INSERT INTO FUNCTION s3
@@ -98,5 +109,5 @@ def test_s3_table_functions_timeouts(started_cluster):
                     )
                 SELECT * FROM numbers(1000000)
             """,
-                settings=settings,
+                settings=timeout_settings,
             )

tests/integration/test_storage_kafka/test_batch_fast.py

@@ -2081,10 +2081,19 @@ def test_kafka_flush_by_time(kafka_cluster, create_query_generator):
 
         cancel = threading.Event()
 
+        # Reuse one producer: `k.kafka_produce` opens a new connection per call,
+        # and a single broker-version probe there can cost seconds, which is
+        # enough to miss the row count asserted below.
+        producer = k.get_kafka_producer(
+            kafka_cluster.kafka_port, k.producer_serializer, retries=15
+        )
+
         def produce():
             while not cancel.is_set():
-                messages = [json.dumps({"key": 0, "value": 0})]
-                k.kafka_produce(kafka_cluster, topic_name, messages)
+                producer.send(
+                    topic=topic_name, value=json.dumps({"key": 0, "value": 0})
+                )
+                producer.flush()
                 time.sleep(0.8)
 
         kafka_thread = threading.Thread(target=produce)
@@ -2102,6 +2111,7 @@ def produce():
 
         cancel.set()
         kafka_thread.join()
+        producer.close()
 
         instance.query(f"""
             DROP TABLE test.{kafka_table}_consumer;

tests/integration/test_storage_kafka/test_batch_slow_0.py

@@ -301,7 +301,6 @@ def test_kafka_formats_with_broken_message(kafka_cluster, create_query_generator
             data_prefix = data_prefix + [""]
         if format_opts.get("printable", False) == False:
             raw_message = "hex(_raw_message)"
-        k.kafka_produce(kafka_cluster, topic_name, data_prefix + data_sample)
         create_query = create_query_generator(
             f"kafka_{format_name}",
             "id Int64, blockNo UInt16, val1 String, val2 Float32, val3 UInt8",
@@ -313,6 +312,10 @@ def test_kafka_formats_with_broken_message(kafka_cluster, create_query_generator
                 "kafka_flush_interval_ms": 1000,
             },
         )
+        # Create both materialized views, then detach/re-attach the Kafka table,
+        # before producing any message. Creating the first view starts the
+        # streaming loop, so producing earlier lets the loop consume and commit
+        # the broken message before the errors view is attached, leaving it empty.
         instance.query(
             f"""
             DROP TABLE IF EXISTS test.kafka_{format_name};
@@ -328,8 +331,12 @@ def test_kafka_formats_with_broken_message(kafka_cluster, create_query_generator
             CREATE MATERIALIZED VIEW test.kafka_errors_{format_name}_mv ENGINE=MergeTree ORDER BY tuple() AS
                 SELECT {raw_message} as raw_message, _error as error, _topic as topic, _partition as partition, _offset as offset FROM test.kafka_{format_name}
                 WHERE length(_error) > 0;
+
+            DETACH TABLE test.kafka_{format_name};
+            ATTACH TABLE test.kafka_{format_name};
             """
         )
+        k.kafka_produce(kafka_cluster, topic_name, data_prefix + data_sample)
 
     raw_expected = """\
 0	0	AM	0.5	1	{topic_name}	0	{offset_0}