ConnectionManager { public void start() { this.creator.start(); long recyleTimeMs = Math.min( poolCleanupPeriodMs, connectionCleanupPeriodMs); LOG.info("Cleaning every {} seconds", TimeUnit.MILLISECONDS.toSeconds(recyleTimeMs)); this.cleaner.scheduleAtFixedRate( new CleanupTask(), 0, recyleTimeMs, TimeUnit.MILLISECONDS); this.running = true; } ConnectionManager(Configuration config); void start(); void close(); ConnectionContext getConnection(UserGroupInformation ugi,
      String nnAddress, Class<?> protocol); int getNumConnectionPools(); int getNumConnections(); int getNumActiveConnections(); int getNumCreatingConnections(); String getJSON();  }
@Test public void testConnectionCreatorWithException() throws Exception { ConnectionPool badPool = new ConnectionPool( conf, UNRESOLVED_TEST_NN_ADDRESS, TEST_USER1, 0, 10, 0.5f, ClientProtocol.class); BlockingQueue<ConnectionPool> queue = new ArrayBlockingQueue<>(1); queue.add(badPool); ConnectionManager.ConnectionCreator connectionCreator = new ConnectionManager.ConnectionCreator(queue); connectionCreator.setDaemon(true); connectionCreator.start(); GenericTestUtils.waitFor(()->queue.isEmpty(), 50, 5000); assertTrue(queue.isEmpty()); assertTrue(connectionCreator.isAlive()); connectionCreator.interrupt(); }