ConnectionManager { public ConnectionContext getConnection(UserGroupInformation ugi, String nnAddress, Class<?> protocol) throws IOException { if (!this.running) { LOG.error( "Cannot get a connection to {} because the manager isn't running", nnAddress); return null; } ConnectionPoolId connectionId = new ConnectionPoolId(ugi, nnAddress, protocol); ConnectionPool pool = null; readLock.lock(); try { pool = this.pools.get(connectionId); } finally { readLock.unlock(); } if (pool == null) { writeLock.lock(); try { pool = this.pools.get(connectionId); if (pool == null) { pool = new ConnectionPool( this.conf, nnAddress, ugi, this.minSize, this.maxSize, this.minActiveRatio, protocol); this.pools.put(connectionId, pool); } } finally { writeLock.unlock(); } } ConnectionContext conn = pool.getConnection(); if (conn == null || !conn.isUsable()) { if (!this.creatorQueue.offer(pool)) { LOG.error("Cannot add more than {} connections at the same time", this.creatorQueueMaxSize); } } if (conn != null && conn.isClosed()) { LOG.error("We got a closed connection from {}", pool); conn = null; } return conn; } 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 testGetConnection() throws Exception { Map<ConnectionPoolId, ConnectionPool> poolMap = connManager.getPools(); final int totalConns = 10; int activeConns = 5; ConnectionPool pool = new ConnectionPool( conf, TEST_NN_ADDRESS, TEST_USER1, 0, 10, 0.5f, ClientProtocol.class); addConnectionsToPool(pool, totalConns, activeConns); poolMap.put( new ConnectionPoolId(TEST_USER1, TEST_NN_ADDRESS, ClientProtocol.class), pool); final int remainingSlots = totalConns - activeConns; for (int i = 0; i < remainingSlots; i++) { ConnectionContext cc = pool.getConnection(); assertTrue(cc.isUsable()); cc.getClient(); activeConns++; } checkPoolConnections(TEST_USER1, totalConns, activeConns); ConnectionContext cc = pool.getConnection(); assertTrue(cc.isActive()); }
@Test public void testValidClientIndex() throws Exception { ConnectionPool pool = new ConnectionPool( conf, TEST_NN_ADDRESS, TEST_USER1, 2, 2, 0.5f, ClientProtocol.class); for(int i = -3; i <= 3; i++) { pool.getClientIndex().set(i); ConnectionContext conn = pool.getConnection(); assertNotNull(conn); assertTrue(conn.isUsable()); } }