diff --git a/tests/integration/test_hedged_requests/test.py b/tests/integration/test_hedged_requests/test.py index 1aafc1aadafa..6c53ef01a5d4 100644 --- a/tests/integration/test_hedged_requests/test.py +++ b/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") diff --git a/tests/integration/test_s3_table_functions/test.py b/tests/integration/test_s3_table_functions/test.py index b2dfddb23ebd..ec565de617ca 100644 --- a/tests/integration/test_s3_table_functions/test.py +++ b/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, ) diff --git a/tests/integration/test_storage_kafka/test_batch_slow_0.py b/tests/integration/test_storage_kafka/test_batch_slow_0.py index 03e7297aa3dd..6f6393f0bd00 100644 --- a/tests/integration/test_storage_kafka/test_batch_slow_0.py +++ b/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} diff --git a/tests/queries/0_stateless/01661_extract_all_groups_throw_fast.sql b/tests/queries/0_stateless/01661_extract_all_groups_throw_fast.sql index afb6866a0df5..f16e83a6d0f5 100644 --- a/tests/queries/0_stateless/01661_extract_all_groups_throw_fast.sql +++ b/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