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

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