'Test to ensure the hosts are marked as down after a OTO is received.
Also to ensure this happens within the expected timeout
@since 3.10
@jira_ticket PYTHON-762
@expected_result all the hosts have been marked as down at some point
@test_category metadata'

def test_heart_beat_timeout(self):

    number_of_dcs = 3
    nodes_per_dc = 100
    query_to_prime = 'INSERT    INTO    test3rf.test    (k,    v)    VALUES    (0,    1);'
    idle_heartbeat_timeout = 5
    idle_heartbeat_interval = 1
    start_and_prime_cluster_defaults(number_of_dcs, nodes_per_dc, CASSANDRA_VERSION)
    self.addCleanup(stop_simulacron)
    listener = TrackDownListener()
    executor = ThreadTracker(max_workers=16)
    cluster = Cluster(compression=False, idle_heartbeat_interval=idle_heartbeat_interval, idle_heartbeat_timeout=idle_heartbeat_timeout, executor_threads=16, execution_profiles={EXEC_PROFILE_DEFAULT: ExecutionProfile(load_balancing_policy=RoundRobinPolicy())})
    self.addCleanup(cluster.shutdown)
    cluster.scheduler.shutdown()
    cluster.executor = executor
    cluster.scheduler = _Scheduler(executor)
    session = cluster.connect(wait_for_all_pools=True)
    cluster.register_listener(listener)
    log = logging.getLogger()
    log.setLevel('CRITICAL')
    self.addCleanup(log.setLevel, 'DEBUG')
    prime_query(query_to_prime, then=NO_THEN)
    futures = []
    for _ in range((number_of_dcs * nodes_per_dc)):
        future = session.execute_async(query_to_prime)
        futures.append(future)
    for f in futures:
        f._event.wait()
        self.assertIsInstance(f._final_exception, OperationTimedOut)
    prime_request(PrimeOptions(then=NO_THEN))
    time.sleep(((idle_heartbeat_timeout + idle_heartbeat_interval) * 2))
    for host in cluster.metadata.all_hosts():
        self.assertIn(host, listener.hosts_marked_down)
    self.assertNotIn('_replace', executor.called_functions)
