@Test public void testBlockingChannelConnectionPool() throws Exception { Properties props = new Properties(); props.put("connectionpool.max.connections.per.port.plain.text", "5"); props.put("connectionpool.max.connections.per.port.ssl", "5"); props.put("clustermap.cluster.name", "test"); props.put("clustermap.datacenter.name", "dc1"); props.put("clustermap.host.name", "localhost"); ConnectionPool connectionPool = new BlockingChannelConnectionPool(new ConnectionPoolConfig(new VerifiableProperties(props)), sslConfig, plainTextClusterMapConfig, new MetricRegistry()); connectionPool.start(); CountDownLatch shouldRelease = new CountDownLatch(1); CountDownLatch releaseComplete = new CountDownLatch(10); AtomicReference<Exception> exception = new AtomicReference<Exception>(); Map<String, CountDownLatch> channelCount = new HashMap<String, CountDownLatch>(); channelCount.put("localhost" + 6667, new CountDownLatch(5)); channelCount.put("localhost" + 6668, new CountDownLatch(5)); channelCount.put("localhost" + 6669, new CountDownLatch(5)); Map<String, Port> channelToPortMap = new HashMap<String, Port>(); channelToPortMap.put("localhost" + 6667, new Port(6667, PortType.PLAINTEXT)); channelToPortMap.put("localhost" + 6668, new Port(6668, PortType.PLAINTEXT)); channelToPortMap.put("localhost" + 6669, new Port(6669, PortType.PLAINTEXT)); for (int i = 0; i < 10; i++) { ConnectionPoolThread connectionPoolThread = new ConnectionPoolThread(channelCount, channelToPortMap, connectionPool, false, shouldRelease, releaseComplete, exception); Thread t = new Thread(connectionPoolThread); t.start(); } for (String channelStr : channelCount.keySet()) { awaitCountdown(channelCount.get(channelStr), 1000, exception, "Timed out waiting for channel count to reach 5"); } for (String channelStr : channelCount.keySet()) { channelCount.put(channelStr, new CountDownLatch(5)); } shouldRelease.countDown(); for (String channelStr : channelCount.keySet()) { awaitCountdown(channelCount.get(channelStr), 1000, exception, "Timed out waiting for channel count to reach 5"); } awaitCountdown(releaseComplete, 2000, exception, "Timed out while waiting for channels to be released"); connectionPool.shutdown(); }
public BlockingChannelConnectionPool(ConnectionPoolConfig config, SSLConfig sslConfig, ClusterMapConfig clusterMapConfig, MetricRegistry registry) throws Exception { connections = new ConcurrentHashMap<String, BlockingChannelInfo>(); this.config = config; this.registry = registry; this.sslConfig = sslConfig; connectionCheckOutTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionCheckOutTime")); connectionCheckInTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionCheckInTime")); connectionDestroyTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionDestroyTime")); totalNumberOfNodesConnectedTo = () -> { int noOfNodesConnectedTo = 0; for (BlockingChannelInfo blockingChannelInfo : connections.values()) { if (blockingChannelInfo.getNumberOfConnections() > 0) { noOfNodesConnectedTo++; } } return noOfNodesConnectedTo; }; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "totalNumberOfNodesConnectedTo"), totalNumberOfNodesConnectedTo); totalNumberOfConnections = () -> { int noOfConnections = 0; for (BlockingChannelInfo blockingChannelInfo : connections.values()) { noOfConnections += blockingChannelInfo.getNumberOfConnections(); } return noOfConnections; }; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "totalNumberOfConnections"), totalNumberOfConnections); requestsWaitingToCheckoutConnectionCount = new AtomicInteger(0); requestsWaitingToCheckoutConnection = requestsWaitingToCheckoutConnectionCount::get; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "requestsWaitingToCheckoutConnection"), requestsWaitingToCheckoutConnection); sslSocketFactoryClientInitializationCount = registry.counter( MetricRegistry.name(BlockingChannelConnectionPool.class, "SslSocketFactoryClientInitializationCount")); sslSocketFactoryClientInitializationErrorCount = registry.counter( MetricRegistry.name(BlockingChannelConnectionPool.class, "SslSocketFactoryClientInitializationErrorCount")); if (clusterMapConfig.clusterMapSslEnabledDatacenters.length() > 0) { initializeSSLSocketFactory(); } else { this.sslSocketFactory = null; } }
BlockingChannelConnectionPool implements ConnectionPool { public BlockingChannelConnectionPool(ConnectionPoolConfig config, SSLConfig sslConfig, ClusterMapConfig clusterMapConfig, MetricRegistry registry) throws Exception { connections = new ConcurrentHashMap<String, BlockingChannelInfo>(); this.config = config; this.registry = registry; this.sslConfig = sslConfig; connectionCheckOutTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionCheckOutTime")); connectionCheckInTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionCheckInTime")); connectionDestroyTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionDestroyTime")); totalNumberOfNodesConnectedTo = () -> { int noOfNodesConnectedTo = 0; for (BlockingChannelInfo blockingChannelInfo : connections.values()) { if (blockingChannelInfo.getNumberOfConnections() > 0) { noOfNodesConnectedTo++; } } return noOfNodesConnectedTo; }; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "totalNumberOfNodesConnectedTo"), totalNumberOfNodesConnectedTo); totalNumberOfConnections = () -> { int noOfConnections = 0; for (BlockingChannelInfo blockingChannelInfo : connections.values()) { noOfConnections += blockingChannelInfo.getNumberOfConnections(); } return noOfConnections; }; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "totalNumberOfConnections"), totalNumberOfConnections); requestsWaitingToCheckoutConnectionCount = new AtomicInteger(0); requestsWaitingToCheckoutConnection = requestsWaitingToCheckoutConnectionCount::get; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "requestsWaitingToCheckoutConnection"), requestsWaitingToCheckoutConnection); sslSocketFactoryClientInitializationCount = registry.counter( MetricRegistry.name(BlockingChannelConnectionPool.class, "SslSocketFactoryClientInitializationCount")); sslSocketFactoryClientInitializationErrorCount = registry.counter( MetricRegistry.name(BlockingChannelConnectionPool.class, "SslSocketFactoryClientInitializationErrorCount")); if (clusterMapConfig.clusterMapSslEnabledDatacenters.length() > 0) { initializeSSLSocketFactory(); } else { this.sslSocketFactory = null; } } }
BlockingChannelConnectionPool implements ConnectionPool { public BlockingChannelConnectionPool(ConnectionPoolConfig config, SSLConfig sslConfig, ClusterMapConfig clusterMapConfig, MetricRegistry registry) throws Exception { connections = new ConcurrentHashMap<String, BlockingChannelInfo>(); this.config = config; this.registry = registry; this.sslConfig = sslConfig; connectionCheckOutTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionCheckOutTime")); connectionCheckInTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionCheckInTime")); connectionDestroyTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionDestroyTime")); totalNumberOfNodesConnectedTo = () -> { int noOfNodesConnectedTo = 0; for (BlockingChannelInfo blockingChannelInfo : connections.values()) { if (blockingChannelInfo.getNumberOfConnections() > 0) { noOfNodesConnectedTo++; } } return noOfNodesConnectedTo; }; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "totalNumberOfNodesConnectedTo"), totalNumberOfNodesConnectedTo); totalNumberOfConnections = () -> { int noOfConnections = 0; for (BlockingChannelInfo blockingChannelInfo : connections.values()) { noOfConnections += blockingChannelInfo.getNumberOfConnections(); } return noOfConnections; }; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "totalNumberOfConnections"), totalNumberOfConnections); requestsWaitingToCheckoutConnectionCount = new AtomicInteger(0); requestsWaitingToCheckoutConnection = requestsWaitingToCheckoutConnectionCount::get; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "requestsWaitingToCheckoutConnection"), requestsWaitingToCheckoutConnection); sslSocketFactoryClientInitializationCount = registry.counter( MetricRegistry.name(BlockingChannelConnectionPool.class, "SslSocketFactoryClientInitializationCount")); sslSocketFactoryClientInitializationErrorCount = registry.counter( MetricRegistry.name(BlockingChannelConnectionPool.class, "SslSocketFactoryClientInitializationErrorCount")); if (clusterMapConfig.clusterMapSslEnabledDatacenters.length() > 0) { initializeSSLSocketFactory(); } else { this.sslSocketFactory = null; } } BlockingChannelConnectionPool(ConnectionPoolConfig config, SSLConfig sslConfig,
      ClusterMapConfig clusterMapConfig, MetricRegistry registry); }
BlockingChannelConnectionPool implements ConnectionPool { public BlockingChannelConnectionPool(ConnectionPoolConfig config, SSLConfig sslConfig, ClusterMapConfig clusterMapConfig, MetricRegistry registry) throws Exception { connections = new ConcurrentHashMap<String, BlockingChannelInfo>(); this.config = config; this.registry = registry; this.sslConfig = sslConfig; connectionCheckOutTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionCheckOutTime")); connectionCheckInTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionCheckInTime")); connectionDestroyTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionDestroyTime")); totalNumberOfNodesConnectedTo = () -> { int noOfNodesConnectedTo = 0; for (BlockingChannelInfo blockingChannelInfo : connections.values()) { if (blockingChannelInfo.getNumberOfConnections() > 0) { noOfNodesConnectedTo++; } } return noOfNodesConnectedTo; }; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "totalNumberOfNodesConnectedTo"), totalNumberOfNodesConnectedTo); totalNumberOfConnections = () -> { int noOfConnections = 0; for (BlockingChannelInfo blockingChannelInfo : connections.values()) { noOfConnections += blockingChannelInfo.getNumberOfConnections(); } return noOfConnections; }; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "totalNumberOfConnections"), totalNumberOfConnections); requestsWaitingToCheckoutConnectionCount = new AtomicInteger(0); requestsWaitingToCheckoutConnection = requestsWaitingToCheckoutConnectionCount::get; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "requestsWaitingToCheckoutConnection"), requestsWaitingToCheckoutConnection); sslSocketFactoryClientInitializationCount = registry.counter( MetricRegistry.name(BlockingChannelConnectionPool.class, "SslSocketFactoryClientInitializationCount")); sslSocketFactoryClientInitializationErrorCount = registry.counter( MetricRegistry.name(BlockingChannelConnectionPool.class, "SslSocketFactoryClientInitializationErrorCount")); if (clusterMapConfig.clusterMapSslEnabledDatacenters.length() > 0) { initializeSSLSocketFactory(); } else { this.sslSocketFactory = null; } } BlockingChannelConnectionPool(ConnectionPoolConfig config, SSLConfig sslConfig,
      ClusterMapConfig clusterMapConfig, MetricRegistry registry); @Override void start(); @Override void shutdown(); @Override ConnectedChannel checkOutConnection(String host, Port port, long timeoutInMs); @Override void checkInConnection(ConnectedChannel connectedChannel); @Override void destroyConnection(ConnectedChannel connectedChannel); }
BlockingChannelConnectionPool implements ConnectionPool { public BlockingChannelConnectionPool(ConnectionPoolConfig config, SSLConfig sslConfig, ClusterMapConfig clusterMapConfig, MetricRegistry registry) throws Exception { connections = new ConcurrentHashMap<String, BlockingChannelInfo>(); this.config = config; this.registry = registry; this.sslConfig = sslConfig; connectionCheckOutTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionCheckOutTime")); connectionCheckInTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionCheckInTime")); connectionDestroyTime = registry.timer(MetricRegistry.name(BlockingChannelConnectionPool.class, "connectionDestroyTime")); totalNumberOfNodesConnectedTo = () -> { int noOfNodesConnectedTo = 0; for (BlockingChannelInfo blockingChannelInfo : connections.values()) { if (blockingChannelInfo.getNumberOfConnections() > 0) { noOfNodesConnectedTo++; } } return noOfNodesConnectedTo; }; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "totalNumberOfNodesConnectedTo"), totalNumberOfNodesConnectedTo); totalNumberOfConnections = () -> { int noOfConnections = 0; for (BlockingChannelInfo blockingChannelInfo : connections.values()) { noOfConnections += blockingChannelInfo.getNumberOfConnections(); } return noOfConnections; }; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "totalNumberOfConnections"), totalNumberOfConnections); requestsWaitingToCheckoutConnectionCount = new AtomicInteger(0); requestsWaitingToCheckoutConnection = requestsWaitingToCheckoutConnectionCount::get; registry.register(MetricRegistry.name(BlockingChannelConnectionPool.class, "requestsWaitingToCheckoutConnection"), requestsWaitingToCheckoutConnection); sslSocketFactoryClientInitializationCount = registry.counter( MetricRegistry.name(BlockingChannelConnectionPool.class, "SslSocketFactoryClientInitializationCount")); sslSocketFactoryClientInitializationErrorCount = registry.counter( MetricRegistry.name(BlockingChannelConnectionPool.class, "SslSocketFactoryClientInitializationErrorCount")); if (clusterMapConfig.clusterMapSslEnabledDatacenters.length() > 0) { initializeSSLSocketFactory(); } else { this.sslSocketFactory = null; } } BlockingChannelConnectionPool(ConnectionPoolConfig config, SSLConfig sslConfig,
      ClusterMapConfig clusterMapConfig, MetricRegistry registry); @Override void start(); @Override void shutdown(); @Override ConnectedChannel checkOutConnection(String host, Port port, long timeoutInMs); @Override void checkInConnection(ConnectedChannel connectedChannel); @Override void destroyConnection(ConnectedChannel connectedChannel); public Gauge<Integer> totalNumberOfConnections; public Gauge<Integer> requestsWaitingToCheckoutConnection; public Counter sslSocketFactoryClientInitializationCount; public Counter sslSocketFactoryClientInitializationErrorCount; }