@Test public void testReplenishConnections() { connectionTracker = new ConnectionTracker(routerConfig.routerScalingUnitMaxConnectionsPerPortPlainText, routerConfig.routerScalingUnitMaxConnectionsPerPortSsl); int minActiveConnectionsCount = 0; int totalConnectionsCount = 0; int availableCount = 0; MockDataNodeId dataNodeId1 = new MockDataNodeId("host1", Collections.singletonList(plainTextPort), Collections.emptyList(), "DC1"); MockDataNodeId dataNodeId2 = new MockDataNodeId("host2", Arrays.asList(plainTextPort, sslPort), Collections.emptyList(), "DC1"); dataNodeId2.setSslEnabledDataCenters(Collections.singletonList("DC1")); connectionTracker.setMinimumActiveConnectionsPercentage(dataNodeId1, 50); minActiveConnectionsCount += 50 * routerConfig.routerScalingUnitMaxConnectionsPerPortPlainText / 100; connectionTracker.setMinimumActiveConnectionsPercentage(dataNodeId2, 200); minActiveConnectionsCount += routerConfig.routerScalingUnitMaxConnectionsPerPortSsl; assertCounts(totalConnectionsCount, availableCount); connectionTracker.replenishConnections(this::mockNewConnection, Integer.MAX_VALUE); totalConnectionsCount += minActiveConnectionsCount; assertCounts(totalConnectionsCount, availableCount); List<String> newConnections = getNewlyEstablishedConnections(); newConnections.forEach(connectionTracker::checkInConnection); availableCount += minActiveConnectionsCount; assertCounts(totalConnectionsCount, availableCount); Assert.assertTrue(connectionTracker.mayCreateNewConnection("host1", plainTextPort, dataNodeId1)); Assert.assertFalse(connectionTracker.mayCreateNewConnection("host2", sslPort, dataNodeId2)); newConnections.stream().limit(2).forEach(connectionTracker::removeConnection); totalConnectionsCount -= 2; availableCount -= 2; assertCounts(totalConnectionsCount, availableCount); connectionTracker.replenishConnections(this::mockNewConnection, Integer.MAX_VALUE); totalConnectionsCount += 2; assertCounts(totalConnectionsCount, availableCount); newConnections = getNewlyEstablishedConnections(); newConnections.forEach(connectionTracker::checkInConnection); availableCount += 2; assertCounts(totalConnectionsCount, availableCount); String conn1 = connectionTracker.checkOutConnection("host1", plainTextPort, dataNodeId1); Assert.assertNotNull(conn1); String conn2 = connectionTracker.checkOutConnection("host2", sslPort, dataNodeId2); Assert.assertNotNull(conn2); availableCount -= 2; assertCounts(totalConnectionsCount, availableCount); connectionTracker.removeConnection(conn1); connectionTracker.checkInConnection(conn2); totalConnectionsCount -= 1; availableCount += 1; assertCounts(totalConnectionsCount, availableCount); connectionTracker.replenishConnections(this::mockNewConnection, Integer.MAX_VALUE); totalConnectionsCount += 1; assertCounts(totalConnectionsCount, availableCount); String connId; while ((connId = connectionTracker.checkOutConnection("host2", sslPort, dataNodeId2)) != null) { connectionTracker.removeConnection(connId); } totalConnectionsCount -= 2; availableCount -= 2; assertCounts(totalConnectionsCount, availableCount); connectionTracker.replenishConnections(this::mockNewConnection, 1); totalConnectionsCount += 1; assertCounts(totalConnectionsCount, availableCount); connectionTracker.replenishConnections(this::mockNewConnection, 1); totalConnectionsCount += 1; assertCounts(totalConnectionsCount, availableCount); }
int replenishConnections(ConnectionFactory connectionFactory, int maxNewConnectionsPerHost) { int newConnections = 0; Iterator<HostPortPoolManager> iter = poolManagersBelowMinActiveConnections.iterator(); while (iter.hasNext()) { HostPortPoolManager poolManager = iter.next(); try { if (poolManager.dataNodeId.getState() == HardwareState.AVAILABLE) { int newConnectionsToHost = 0; while (newConnectionsToHost < maxNewConnectionsPerHost && !poolManager.hasMinActiveConnections()) { String connId = connectionFactory.connect(poolManager.host, poolManager.port); poolManager.incrementPoolCount(); connectionIdToPoolManager.put(connId, poolManager); totalManagedConnectionsCount++; newConnections++; newConnectionsToHost++; } if (poolManager.hasMinActiveConnections()) { iter.remove(); } } } catch (IOException e) { LOGGER.warn("Encountered exception while replenishing connections to {}:{}.", poolManager.host, poolManager.port.getPort(), e); } } return newConnections; }
ConnectionTracker { int replenishConnections(ConnectionFactory connectionFactory, int maxNewConnectionsPerHost) { int newConnections = 0; Iterator<HostPortPoolManager> iter = poolManagersBelowMinActiveConnections.iterator(); while (iter.hasNext()) { HostPortPoolManager poolManager = iter.next(); try { if (poolManager.dataNodeId.getState() == HardwareState.AVAILABLE) { int newConnectionsToHost = 0; while (newConnectionsToHost < maxNewConnectionsPerHost && !poolManager.hasMinActiveConnections()) { String connId = connectionFactory.connect(poolManager.host, poolManager.port); poolManager.incrementPoolCount(); connectionIdToPoolManager.put(connId, poolManager); totalManagedConnectionsCount++; newConnections++; newConnectionsToHost++; } if (poolManager.hasMinActiveConnections()) { iter.remove(); } } } catch (IOException e) { LOGGER.warn("Encountered exception while replenishing connections to {}:{}.", poolManager.host, poolManager.port.getPort(), e); } } return newConnections; } }
ConnectionTracker { int replenishConnections(ConnectionFactory connectionFactory, int maxNewConnectionsPerHost) { int newConnections = 0; Iterator<HostPortPoolManager> iter = poolManagersBelowMinActiveConnections.iterator(); while (iter.hasNext()) { HostPortPoolManager poolManager = iter.next(); try { if (poolManager.dataNodeId.getState() == HardwareState.AVAILABLE) { int newConnectionsToHost = 0; while (newConnectionsToHost < maxNewConnectionsPerHost && !poolManager.hasMinActiveConnections()) { String connId = connectionFactory.connect(poolManager.host, poolManager.port); poolManager.incrementPoolCount(); connectionIdToPoolManager.put(connId, poolManager); totalManagedConnectionsCount++; newConnections++; newConnectionsToHost++; } if (poolManager.hasMinActiveConnections()) { iter.remove(); } } } catch (IOException e) { LOGGER.warn("Encountered exception while replenishing connections to {}:{}.", poolManager.host, poolManager.port.getPort(), e); } } return newConnections; } ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl); }
ConnectionTracker { int replenishConnections(ConnectionFactory connectionFactory, int maxNewConnectionsPerHost) { int newConnections = 0; Iterator<HostPortPoolManager> iter = poolManagersBelowMinActiveConnections.iterator(); while (iter.hasNext()) { HostPortPoolManager poolManager = iter.next(); try { if (poolManager.dataNodeId.getState() == HardwareState.AVAILABLE) { int newConnectionsToHost = 0; while (newConnectionsToHost < maxNewConnectionsPerHost && !poolManager.hasMinActiveConnections()) { String connId = connectionFactory.connect(poolManager.host, poolManager.port); poolManager.incrementPoolCount(); connectionIdToPoolManager.put(connId, poolManager); totalManagedConnectionsCount++; newConnections++; newConnectionsToHost++; } if (poolManager.hasMinActiveConnections()) { iter.remove(); } } } catch (IOException e) { LOGGER.warn("Encountered exception while replenishing connections to {}:{}.", poolManager.host, poolManager.port.getPort(), e); } } return newConnections; } ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl);  }
ConnectionTracker { int replenishConnections(ConnectionFactory connectionFactory, int maxNewConnectionsPerHost) { int newConnections = 0; Iterator<HostPortPoolManager> iter = poolManagersBelowMinActiveConnections.iterator(); while (iter.hasNext()) { HostPortPoolManager poolManager = iter.next(); try { if (poolManager.dataNodeId.getState() == HardwareState.AVAILABLE) { int newConnectionsToHost = 0; while (newConnectionsToHost < maxNewConnectionsPerHost && !poolManager.hasMinActiveConnections()) { String connId = connectionFactory.connect(poolManager.host, poolManager.port); poolManager.incrementPoolCount(); connectionIdToPoolManager.put(connId, poolManager); totalManagedConnectionsCount++; newConnections++; newConnectionsToHost++; } if (poolManager.hasMinActiveConnections()) { iter.remove(); } } } catch (IOException e) { LOGGER.warn("Encountered exception while replenishing connections to {}:{}.", poolManager.host, poolManager.port.getPort(), e); } } return newConnections; } ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl);   }