@Test public void testConnectionTracker() throws IOException { connectionTracker = new ConnectionTracker(routerConfig.routerScalingUnitMaxConnectionsPerPortPlainText, routerConfig.routerScalingUnitMaxConnectionsPerPortSsl); int totalConnectionsCount = 0; int availableCount = 0; List<String> mountPaths = Collections.singletonList("/mnt/path"); MockDataNodeId dataNodeId1 = new MockDataNodeId("host1", Collections.singletonList(plainTextPort), mountPaths, "DC1"); MockDataNodeId dataNodeId2 = new MockDataNodeId("host2", Arrays.asList(plainTextPort, sslPort), mountPaths, "DC1"); boolean done = false; do { String connId = connectionTracker.checkOutConnection("host1", plainTextPort, dataNodeId1); if (connId == null && connectionTracker.mayCreateNewConnection("host1", plainTextPort, dataNodeId1)) { connId = connectionTracker.connectAndTrack(this::mockNewConnection, "host1", plainTextPort, dataNodeId1); } else { done = true; } } while (!done); totalConnectionsCount = routerConfig.routerScalingUnitMaxConnectionsPerPortPlainText; assertCounts(totalConnectionsCount, availableCount); done = false; do { String connId = connectionTracker.checkOutConnection("host2", sslPort, dataNodeId2); if (connId == null && connectionTracker.mayCreateNewConnection("host2", sslPort, dataNodeId2)) { connId = connectionTracker.connectAndTrack(this::mockNewConnection, "host2", sslPort, dataNodeId1); } else { done = true; } } while (!done); totalConnectionsCount += routerConfig.routerScalingUnitMaxConnectionsPerPortSsl; assertCounts(totalConnectionsCount, availableCount); for (String conn : getNewlyEstablishedConnections()) { connectionTracker.checkInConnection(conn); availableCount++; } assertCounts(totalConnectionsCount, availableCount); String conn = connectionTracker.checkOutConnection("host2", sslPort, dataNodeId2); Assert.assertNotNull(conn); availableCount--; assertCounts(totalConnectionsCount, availableCount); connectionTracker.checkInConnection(conn); availableCount++; assertCounts(totalConnectionsCount, availableCount); String checkedInConn = conn; Set<String> connIdSet = new HashSet<String>(); for (int i = 0; i < routerConfig.routerScalingUnitMaxConnectionsPerPortSsl; i++) { conn = connectionTracker.checkOutConnection("host2", sslPort, dataNodeId2); Assert.assertNotNull(conn); connIdSet.add(conn); availableCount--; } assertCounts(totalConnectionsCount, availableCount); Assert.assertTrue(connIdSet.contains(checkedInConn)); Assert.assertNull(connectionTracker.checkOutConnection("host2", sslPort, dataNodeId2)); assertCounts(totalConnectionsCount, availableCount); try { connectionTracker.checkInConnection("invalid"); Assert.fail("Invalid connections should not get checked in."); } catch (IllegalArgumentException e) { } try { connectionTracker.removeConnection("invalid"); Assert.fail("Removing invalid connections should not succeed."); } catch (IllegalArgumentException e) { } String conn11 = connectionTracker.checkOutConnection("host1", plainTextPort, dataNodeId1); Assert.assertNotNull(conn11); availableCount--; String conn12 = connectionTracker.checkOutConnection("host1", plainTextPort, dataNodeId1); Assert.assertNotNull(conn12); availableCount--; connectionTracker.checkInConnection(conn11); availableCount++; connectionTracker.removeConnection(conn12); totalConnectionsCount--; connectionTracker.removeConnection(conn11); totalConnectionsCount--; availableCount--; try { connectionTracker.removeConnection(conn11); Assert.fail("Removing the same connection twice should not succeed."); } catch (IllegalArgumentException e) { } assertCounts(totalConnectionsCount, availableCount); Assert.assertNotNull(connectionTracker.checkOutConnection("host1", plainTextPort, dataNodeId1)); availableCount--; for (int i = 0; i < 2; i++) { Assert.assertNull("There should not be any available connections to check out", connectionTracker.checkOutConnection("host1", plainTextPort, dataNodeId1)); Assert.assertTrue("It should be okay to initiate a new connection", connectionTracker.mayCreateNewConnection("host1", plainTextPort, dataNodeId1)); connectionTracker.connectAndTrack(this::mockNewConnection, "host1", plainTextPort, dataNodeId1); totalConnectionsCount++; } Assert.assertNull("There should not be any available connections to check out", connectionTracker.checkOutConnection("host1", plainTextPort, dataNodeId1)); Assert.assertFalse("It should not be okay to initiate a new connection", connectionTracker.mayCreateNewConnection("host1", plainTextPort, dataNodeId1)); assertCounts(totalConnectionsCount, availableCount); }
ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl) { this.maxConnectionsPerPortPlainText = maxConnectionsPerPortPlainText; this.maxConnectionsPerPortSsl = maxConnectionsPerPortSsl; }
ConnectionTracker { ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl) { this.maxConnectionsPerPortPlainText = maxConnectionsPerPortPlainText; this.maxConnectionsPerPortSsl = maxConnectionsPerPortSsl; } }
ConnectionTracker { ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl) { this.maxConnectionsPerPortPlainText = maxConnectionsPerPortPlainText; this.maxConnectionsPerPortSsl = maxConnectionsPerPortSsl; } ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl); }
ConnectionTracker { ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl) { this.maxConnectionsPerPortPlainText = maxConnectionsPerPortPlainText; this.maxConnectionsPerPortSsl = maxConnectionsPerPortSsl; } ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl);  }
ConnectionTracker { ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl) { this.maxConnectionsPerPortPlainText = maxConnectionsPerPortPlainText; this.maxConnectionsPerPortSsl = maxConnectionsPerPortSsl; } ConnectionTracker(int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl);   }