ConnectionPool implements ClusterNodeStatusDetailListener { public <T> Connection<T> reserveConnection(Class<T> serviceInterface, RNodeAddress addr, SocketListener socketListener) throws ConnectionException, InterruptedException { DiqubeThriftServiceInfo<T> serviceInfo = diqubeThriftServiceInfoManager.getServiceInfo(serviceInterface); Connection<T> res = null; UUID executionUuid = QueryUuid.getCurrentExecutionUuid(); if (executionUuid != null && openConnectionsByExecutionUuid.containsKey(executionUuid)) { logger.trace("Execution {} has connections opened already, not blocking this thread", executionUuid); res = reserveAvailableOrCreateConnection(serviceInfo, addr, socketListener); } if (res == null) { res = reserveAvailableConnection(serviceInfo, addr, socketListener); if (res == null) { if (overallOpenConnections.get() >= Math.round(earlyCloseLevel * connectionSoftLimit)) { while (!connectionTimeouts.isEmpty()) if (timeoutCloseConnectionIfViable(() -> connectionTimeouts.pollFirstEntry())) break; } res = reserveAvailableConnection(serviceInfo, addr, socketListener); } while (res == null && overallOpenConnections.get() >= connectionSoftLimit && (executionUuid == null || !openConnectionsByExecutionUuid.containsKey(executionUuid))) { logger.debug("Blocking thread as there are no connections available ({}/{}).", overallOpenConnections.get(), connectionSoftLimit); synchronized (connectionsAvailableWait) { connectionsAvailableWait.wait(1000); } res = reserveAvailableConnection(serviceInfo, addr, socketListener); } if (res == null) res = reserveAvailableOrCreateConnection(serviceInfo, addr, socketListener); } if (executionUuid != null) { AtomicInteger openConnCount = openConnectionsByExecutionUuid.get(executionUuid); if (openConnCount == null) { synchronized (executionUuid) { if (!openConnectionsByExecutionUuid.containsKey(executionUuid)) openConnectionsByExecutionUuid.put(executionUuid, new AtomicInteger(0)); openConnCount = openConnectionsByExecutionUuid.get(executionUuid); } } synchronized (executionUuid) { int openConns = openConnCount.incrementAndGet(); logger.trace("Execution {} has {} open connections", executionUuid, openConns); if (openConnectionsByExecutionUuid.get(executionUuid) != openConnCount) openConnectionsByExecutionUuid.put(executionUuid, openConnCount); } } res.setExecutionUuid(executionUuid); return res; }  @PostConstruct void initialize(); @PreDestroy void cleanup(); Connection<T> reserveConnection(Class<T> serviceInterface, RNodeAddress addr,
      SocketListener socketListener); void releaseConnection(Connection<T> connection); @Override void nodeDied(RNodeAddress nodeAddr); @Override void nodeAlive(RNodeAddress nodeAddr);  }
@Test public void connectionReuse() throws ConnectionException, InterruptedException, IOException { initPool(Integer.MAX_VALUE, 2, Integer.MAX_VALUE, .95); TestConnection<ClusterManagementService.Iface> conn = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR1, null); int connId = conn.getId(); Assert.assertEquals(conn.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); conn.close(); conn = (TestConnection<ClusterManagementService.Iface>) pool.reserveConnection(ClusterManagementService.Iface.class, ADDR1, null); conn.close(); Assert.assertEquals(conn.getId(), connId, "Expected that connection is re-used"); Assert.assertEquals(conn.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); }
@Test public void connectionUsableAfterReserve() throws ConnectionException, InterruptedException, IOException { initPool(Integer.MAX_VALUE, 2, Integer.MAX_VALUE, .95); TestConnection<ClusterManagementService.Iface> conn = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR1, null); conn.getService(); }
@Test public void blockTimeoutNewConnectionHighEarlyCloseLevel() throws ConnectionException, InterruptedException, IOException { initPool(Integer.MAX_VALUE, 1, 3_000, 2); TestConnection<ClusterManagementService.Iface> conn = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR1, null); int connId = conn.getId(); Assert.assertEquals(conn.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); long beforeNanos = System.nanoTime(); conn.close(); TestConnection<ClusterManagementService.Iface> conn2 = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR2, null); long afterNanos = System.nanoTime(); conn2.close(); Mockito.verify(conn.getTransport()).close(); Assert.assertTrue(afterNanos - beforeNanos >= 3_000_000_000L, "Expected to have blocked at least the timeout time until we receive a new connection, but waited only " + (afterNanos - beforeNanos) + " nanos"); Assert.assertNotEquals(conn2.getId(), connId, "Expected that connection was not re-used, as first " + "connection should have been closed before the second was opened"); Assert.assertEquals(conn2.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); }
@Test public void earlyCloseOfAvailableConnections() throws ConnectionException, InterruptedException, IOException { initPool(Integer.MAX_VALUE, 1, Integer.MAX_VALUE, .95); TestConnection<ClusterManagementService.Iface> conn = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR1, null); int connId = conn.getId(); Assert.assertEquals(conn.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); conn.close(); TestConnection<ClusterManagementService.Iface> conn2 = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR2, null); conn2.close(); Mockito.verify(conn.getTransport()).close(); Assert.assertNotEquals(conn2.getId(), connId, "Expected that connection was not re-used, as first " + "connection should have been closed before the second was opened"); Assert.assertEquals(conn2.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); }
@Test public void blockKeepAliveDeadNewConnectionHighEarlyCloseLevel() throws ConnectionException, InterruptedException, IOException { initPool(1_000, 1, Integer.MAX_VALUE, 2); TestConnection<ClusterManagementService.Iface> conn = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR1, null); int connId = conn.getId(); Assert.assertEquals(conn.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); long beforeNanos = System.nanoTime(); conFac.adjustMorphOfConn(connId, new Consumer<TestConnection<?>>() { @Override public void accept(TestConnection<?> t) { try { if (t.getServiceInfo().getServiceInterface().equals(KeepAliveService.Iface.class)) { t.pooledCAS(true, false); Mockito.doThrow(TException.class).when(((KeepAliveService.Iface) t.getService())).ping(); t.pooledCAS(false, true); } } catch (TException e) { throw new RuntimeException(e); } } }); conn.close(); TestConnection<ClusterManagementService.Iface> conn2 = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR2, null); long afterNanos = System.nanoTime(); conn2.close(); Mockito.verify(conn.getTransport()).close(); Assert.assertTrue(afterNanos - beforeNanos >= 1_000_000_000L, "Expected to have blocked at least the keep-alive time until it was found that the first conn is dead and a " + "new conn was opened, but waited only " + (afterNanos - beforeNanos) + " nanos"); Assert.assertNotEquals(conn2.getId(), connId, "Expected that connection was not re-used, as " + "first connection should have been closed before the second was opened"); Assert.assertEquals(conn2.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); }
@Test public void reuseDeadNewConnection() throws ConnectionException, InterruptedException, IOException { initPool(Integer.MAX_VALUE, 1, Integer.MAX_VALUE, .95); TestConnection<ClusterManagementService.Iface> conn = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR1, null); int connId = conn.getId(); Assert.assertEquals(conn.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); conFac.adjustMorphOfConn(connId, new Consumer<TestConnection<?>>() { @Override public void accept(TestConnection<?> t) { try { if (t.getServiceInfo().getServiceInterface().equals(KeepAliveService.Iface.class)) { t.pooledCAS(true, false); Mockito.doThrow(TException.class).when(((KeepAliveService.Iface) t.getService())).ping(); t.pooledCAS(false, true); } } catch (TException e) { throw new RuntimeException(e); } } }); conn.close(); TestConnection<ClusterManagementService.Iface> conn2 = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR1, null); conn2.close(); Mockito.verify(conn.getTransport()).close(); Assert.assertNotEquals(conn2.getId(), connId, "Expected that connection was not re-used, as " + "first connection was tried to be re-used but appeared to have died"); Assert.assertEquals(conn2.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); }
@Test public void executionGetsAnotherConn() throws ConnectionException, InterruptedException, IOException { initPool(Integer.MAX_VALUE, 1, Integer.MAX_VALUE, .95); try { QueryUuid.setCurrentQueryUuidAndExecutionUuid(UUID.randomUUID(), UUID.randomUUID()); TestConnection<ClusterManagementService.Iface> conn = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR1, null); int connId = conn.getId(); Assert.assertEquals(conn.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); TestConnection<ClusterManagementService.Iface> conn2 = (TestConnection<ClusterManagementService.Iface>) pool .reserveConnection(ClusterManagementService.Iface.class, ADDR2, null); conn2.close(); conn.close(); Assert.assertNotEquals(conn2.getId(), connId, "Expected that connection was not re-used, as both connections should be open at the same time"); Assert.assertEquals(conn2.getServiceInfo().getServiceInterface(), ClusterManagementService.Iface.class, "Correct service expected"); } finally { QueryUuid.clearCurrent(); } }