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_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_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}

tests/integration/test_tcp_handler_connection_limits/test.py

@@ -1,6 +1,5 @@
 import pytest
 import subprocess
-import time
 from helpers.cluster import ClickHouseCluster
 
 cluster = ClickHouseCluster(__file__)
@@ -14,11 +13,6 @@ def started_cluster():
     finally:
         cluster.shutdown()
 
-@pytest.fixture(scope="module", autouse=True)
-def stabilize_container(started_cluster):
-    """Wait for container startup processes to complete before running tests"""
-    time.sleep(1)
-
 def execute_queries_persistent_connection(queries):
     """Execute multiple queries through a single persistent clickhouse-client connection"""
     proc = subprocess.Popen(
@@ -34,17 +28,19 @@ def execute_queries_persistent_connection(queries):
 
     return stdout, stderr
 
-def get_connection_done_count():
-    try:
-        log_result = node.exec_in_container(
-            ["grep", "-c", "Done processing connection", "/var/log/clickhouse-server/clickhouse-server.log"]
-        )
-        return int(log_result.strip())
-    except Exception:
-        return 0
+def get_limit_closed_count(reason):
+    """Count the connections that the server closed because a limit was reached.
+
+    Counting every closed connection instead would be racy: the readiness probe of
+    `cluster.start` connects to the port and closes it without sending any data, and the
+    server accepts that connection only once it starts serving, which can happen after the
+    test has already sampled the initial count. Connections closed for other reasons never
+    report a limit, so counting only those keeps the assertion exact.
+    """
+    return int(node.count_in_log(f"Closing connection due to limits: {reason}").strip())
 
 def test_query_count_limit(started_cluster):
-    initial_count = get_connection_done_count()
+    initial_count = get_limit_closed_count("queries=")
 
     queries = ["SELECT 1;", "SELECT 2;", "SELECT 3;", "SELECT 4;", "SELECT 5;"]
     stdout, stderr = execute_queries_persistent_connection(queries)
@@ -53,17 +49,17 @@ def test_query_count_limit(started_cluster):
     assert "4" not in stdout and "5" not in stdout
     assert "TCP_CONNECTION_LIMIT_REACHED" in stderr
 
-    final_count = get_connection_done_count()
+    final_count = get_limit_closed_count("queries=")
     assert final_count == initial_count + 1, f"Expected exactly 1 connection closure, got {final_count - initial_count}"
 
 def test_time_limit(started_cluster):
-    initial_count = get_connection_done_count()
+    initial_count = get_limit_closed_count("elapsed=")
 
     queries = ["SELECT sleep(3);", "SELECT 1;", "SELECT 2;"]
     stdout, stderr = execute_queries_persistent_connection(queries)
 
     assert "1" not in stdout and "2" not in stdout
     assert "TCP_CONNECTION_LIMIT_REACHED" in stderr
 
-    final_count = get_connection_done_count()
+    final_count = get_limit_closed_count("elapsed=")
     assert final_count == initial_count + 1, f"Expected exactly 1 connection closure, got {final_count - initial_count}"

tests/queries/0_stateless/01184_long_insert_values_huge_strings.sh

@@ -17,7 +17,8 @@ done;
 wait
 
 $CLICKHOUSE_CLIENT -q "select count() from huge_strings"
-$CLICKHOUSE_CLIENT -q "select sum(l = length(s)) from huge_strings"
-$CLICKHOUSE_CLIENT -q "select sum(h = cityHash64(s)) from huge_strings"
+# Pin `max_threads`: each read stream holds its own buffer for a ~9 MB row of `s`, so the randomized 32-thread draw exceeds `max_memory_usage`.
+$CLICKHOUSE_CLIENT -q "select sum(l = length(s)) from huge_strings SETTINGS max_threads = 3"
+$CLICKHOUSE_CLIENT -q "select sum(h = cityHash64(s)) from huge_strings SETTINGS max_threads = 3"
 
 $CLICKHOUSE_CLIENT -q "drop table huge_strings"

tests/queries/0_stateless/01661_extract_all_groups_throw_fast.sql

@@ -1,2 +1,4 @@
-SELECT repeat('abcdefghijklmnopqrstuvwxyz', number * 100) AS haystack, extractAllGroupsHorizontal(haystack, '(\\w)') AS matches FROM numbers(1023); -- { serverError TOO_LARGE_ARRAY_SIZE }
+-- 1001 matches in one row exceeds `regexp_max_matches_per_row` (default 1000) and trips the fast-throw.
+-- The low `max_memory_usage` also guards against regressing to a huge fixture that could hit MEMORY_LIMIT_EXCEEDED first.
+SELECT extractAllGroupsHorizontal(materialize(repeat('a', 1001)), '(\\w)') FORMAT Null SETTINGS max_memory_usage = 100000000; -- { serverError TOO_LARGE_ARRAY_SIZE }
 SELECT count(extractAllGroupsHorizontal(materialize('a'), '(a)')) FROM numbers(1000000) FORMAT Null; -- shouldn't fail

tests/queries/0_stateless/02040_clickhouse_benchmark_query_id_pass_through.sh

@@ -6,6 +6,11 @@ CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 
 query_id="${CLICKHOUSE_DATABASE}_$$"
 benchmark_args=(
+    # A loaded runner can take longer than the default 10 s handshake_timeout_ms
+    # to send Hello; the benchmark then exits without running a query and
+    # query_log has 0 rows instead of 3.
+    --connect_timeout 60
+    --handshake_timeout_ms 60000
     --iterations 1
     --log_queries 1
     --query_id "$query_id"