public void start() {
    // Start the thread that creates connections asynchronously
    this.creator.start();

    // Schedule a task to remove stale connection pools and sockets
    long recycleTimeMs = Math.min(
        poolCleanupPeriodMs, connectionCleanupPeriodMs);
    LOG.info("Cleaning every {} seconds",
        TimeUnit.MILLISECONDS.toSeconds(recycleTimeMs));
    this.cleaner.scheduleAtFixedRate(
        new CleanupTask(), 0, recycleTimeMs, TimeUnit.MILLISECONDS);

    // Mark the manager as running
    this.running = true;
  }
@Test
  public void testConnectionCreatorWithException() throws Exception {
    // Create a bad connection pool pointing to unresolvable namenode address.
    ConnectionPool badPool = new ConnectionPool(
        conf, UNRESOLVED_TEST_NN_ADDRESS, TEST_USER1, 0, 10, 0.5f,
        ClientProtocol.class, null);
    BlockingQueue<ConnectionPool> queue = new ArrayBlockingQueue<>(1);
    queue.add(badPool);
    ConnectionManager.ConnectionCreator connectionCreator =
        new ConnectionManager.ConnectionCreator(queue);
    connectionCreator.setDaemon(true);
    connectionCreator.start();
    // Wait to make sure async thread is scheduled and picks
    GenericTestUtils.waitFor(queue::isEmpty, 50, 5000);
    // At this point connection creation task should be definitely picked up.
    assertTrue(queue.isEmpty());
    // At this point connection thread should still be alive.
    assertTrue(connectionCreator.isAlive());
    // Stop the thread as test is successful at this point
    connectionCreator.interrupt();
  }