def test_with_replicated_merge_tree_multithread():
        thread_num = 15
        table_name = "async_insert_replicated_mt_multithread_table"
    
        create_query = " ".join(
            (
                "CREATE TABLE {} (a UInt64, b Array(UInt64))".format(table_name),
                "ENGINE=ReplicatedMergeTree('/clickhouse/tables/test/{}', 'node')".format(
                    table_name
                ),
                "ORDER BY a",
            )
        )
    
        node.query(create_query)
    
>       _insert_queries_in_parallel(
            table_name,
            _query_settings,
            thread_num=15,
            tasks=100,
            max_values_size=1000,
            array_size_range=[10, 15],
        )

test_async_insert_adaptive_busy_timeout/test.py:179: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
test_async_insert_adaptive_busy_timeout/test.py:83: in _insert_queries_in_parallel
    p.starmap(
/usr/lib/python3.10/multiprocessing/pool.py:375: in starmap
    return self._map_async(func, iterable, starmapstar, chunksize).get()
/usr/lib/python3.10/multiprocessing/pool.py:768: in get
    self.wait(timeout)
/usr/lib/python3.10/multiprocessing/pool.py:765: in wait
    self._event.wait(timeout)
/usr/lib/python3.10/threading.py:607: in wait
    signaled = self._cond.wait(timeout)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Condition(<unlocked _thread.lock object at 0xff18c028a880>, 0)>
timeout = None

    def wait(self, timeout=None):
        """Wait until notified or until a timeout occurs.
    
        If the calling thread has not acquired the lock when this method is
        called, a RuntimeError is raised.
    
        This method releases the underlying lock, and then blocks until it is
        awakened by a notify() or notify_all() call for the same condition
        variable in another thread, or until the optional timeout occurs. Once
        awakened or timed out, it re-acquires the lock and returns.
    
        When the timeout argument is present and not None, it should be a
        floating point number specifying a timeout for the operation in seconds
        (or fractions thereof).
    
        When the underlying lock is an RLock, it is not released using its
        release() method, since this may not actually unlock the lock when it
        was acquired multiple times recursively. Instead, an internal interface
        of the RLock class is used, which really unlocks it even when it has
        been recursively acquired several times. Another internal interface is
        then used to restore the recursion level when the lock is reacquired.
    
        """
        if not self._is_owned():
            raise RuntimeError("cannot wait on un-acquired lock")
        waiter = _allocate_lock()
        waiter.acquire()
        self._waiters.append(waiter)
        saved_state = self._release_save()
        gotit = False
        try:    # restore state no matter what (e.g., KeyboardInterrupt)
            if timeout is None:
>               waiter.acquire()
E               Failed: Timeout >900.0s