Merge pull request #33164 from azat/fix-test_async_drain_connection

Fix test_async_drain_connection flakiness

tests/integration/test_async_drain_connection/test.py

@@ -1,21 +1,21 @@
-import os
-import sys
-import time
-from multiprocessing.dummy import Pool
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+
 import pytest
 from helpers.cluster import ClickHouseCluster
 
 cluster = ClickHouseCluster(__file__)
-node = cluster.add_instance("node", main_configs=["configs/config.xml"])
+node = cluster.add_instance('node', main_configs=['configs/config.xml'])
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture(scope='module')
 def started_cluster():
     try:
         cluster.start()
-        node.query(
-            'create table t (number UInt64) engine = Distributed(test_cluster_two_shards, system, numbers);'
-        )
+        node.query("""
+        create table t (number UInt64)
+        engine = Distributed(test_cluster_two_shards, system, numbers)
+        """)
         yield cluster
 
     finally:
@@ -23,14 +23,14 @@ def started_cluster():
 
 
 def test_filled_async_drain_connection_pool(started_cluster):
-    busy_pool = Pool(10)
-
-    def execute_query(i):
+    def execute_queries(_):
         for _ in range(100):
-            node.query('select * from t where number = 0 limit 2;',
-                       settings={
-                           "sleep_in_receive_cancel_ms": 10000000,
-                           "max_execution_time": 5
-                       })
+            node.query('select * from t where number = 0 limit 2', settings={
+                'sleep_in_receive_cancel_ms': int(10e6),
+                'max_execution_time': 5,
+                # decrease drain_timeout to make test more stable
+                # (another way is to increase max_execution_time, but this will make test slower)
+                'drain_timeout': 1,
+            })
 
-    p = busy_pool.map(execute_query, range(10))
+    any(map(execute_queries, range(10)))