lock.assertHeld()

    task.nextExecuteNanoTime = -1L
    val queue = task.queue!!
    queue.futureTasks.remove(task)
    readyQueues.remove(queue)
    queue.activeTask = task
    busyQueues.add(queue)
  }

  private fun runTask(task: Task) {
    val currentThread = Thread.currentThread()
    val oldName = currentThread.name
    currentThread.name = task.name

    var delayNanos = -1L
    try {

    assertThat(pool.closeConnections(100L)).isEqualTo(50L)
    assertThat(pool.connectionCount()).isEqualTo(2)
    assertThat(c1.socket().isClosed).isFalse()
    assertThat(c2.socket().isClosed).isFalse()

    // Add a third connection
    val c3 = factory.newConnection(pool, routeC1, 75L)

    // The third connection bounces the first.
    assertThat(pool.closeConnections(100L)).isEqualTo(0L)

  private val taskRunner = taskFaker.taskRunner

  /**
   * Note that we don't use the same [TaskFaker] for this factory. That way off-topic tasks like
   * connection pool maintenance tasks don't add noise to route planning tests.
   */
  private val routePlanner = FakeRoutePlanner(taskFaker = taskFaker)
  private val finder = FastFallbackExchangeFinder(routePlanner, taskRunner)

  @AfterEach
  fun tearDown() {

      object : Task("task one") {
        val schedules = mutableListOf(200.µs)

        override fun runOnce(): Long {
          if (schedules.isNotEmpty()) {
            redQueue.schedule(this, schedules.removeAt(0)) // Add it at the end also.
          }
          log += "scheduledTasks=${redQueue.scheduledTasks}"
          return -1L
        }
      }
    redQueue.schedule(task, 100.µs)