ConnectionManager { @VisibleForTesting void cleanup(ConnectionPool pool) { if (pool.getNumConnections() > pool.getMinSize()) { long timeSinceLastActive = Time.now() - pool.getLastActiveTime(); int total = pool.getNumConnections(); int active = pool.getNumActiveConnections(); float poolMinActiveRatio = pool.getMinActiveRatio(); if (timeSinceLastActive > connectionCleanupPeriodMs || active < poolMinActiveRatio * total) { List<ConnectionContext> conns = pool.removeConnections(1); for (ConnectionContext conn : conns) { conn.close(); } LOG.debug("Removed connection {} used {} seconds ago. " + "Pool has {}/{} connections", pool.getConnectionPoolId(), TimeUnit.MILLISECONDS.toSeconds(timeSinceLastActive), pool.getNumConnections(), pool.getMaxSize()); } } } 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 testCleanup() throws Exception { Map<ConnectionPoolId, ConnectionPool> poolMap = connManager.getPools(); ConnectionPool pool1 = new ConnectionPool( conf, TEST_NN_ADDRESS, TEST_USER1, 0, 10, 0.5f, ClientProtocol.class); addConnectionsToPool(pool1, 9, 4); poolMap.put( new ConnectionPoolId(TEST_USER1, TEST_NN_ADDRESS, ClientProtocol.class), pool1); ConnectionPool pool2 = new ConnectionPool( conf, TEST_NN_ADDRESS, TEST_USER2, 0, 10, 0.5f, ClientProtocol.class); addConnectionsToPool(pool2, 10, 10); poolMap.put( new ConnectionPoolId(TEST_USER2, TEST_NN_ADDRESS, ClientProtocol.class), pool2); checkPoolConnections(TEST_USER1, 9, 4); checkPoolConnections(TEST_USER2, 10, 10); connManager.cleanup(pool1); checkPoolConnections(TEST_USER1, 8, 4); checkPoolConnections(TEST_USER2, 10, 10); connManager.cleanup(pool1); checkPoolConnections(TEST_USER1, 8, 4); checkPoolConnections(TEST_USER2, 10, 10); ConnectionPool pool3 = new ConnectionPool( conf, TEST_NN_ADDRESS, TEST_USER3, 2, 10, 0.5f, ClientProtocol.class); addConnectionsToPool(pool3, 8, 0); poolMap.put( new ConnectionPoolId(TEST_USER3, TEST_NN_ADDRESS, ClientProtocol.class), pool3); checkPoolConnections(TEST_USER3, 10, 0); for (int i = 0; i < 10; i++) { connManager.cleanup(pool3); } checkPoolConnections(TEST_USER3, 2, 0); addConnectionsToPool(pool3, 8, 2); checkPoolConnections(TEST_USER3, 10, 2); for (int i = 0; i < 10; i++) { connManager.cleanup(pool3); } checkPoolConnections(TEST_USER3, 4, 2); }