@Test public void testWarmUpConnectionsFailedAll() { selector.setState(MockSelectorState.FailConnectionInitiationOnPoll); List<ResponseInfo> responseInfos = new ArrayList<>(); Assert.assertEquals("Connection count is not expected", 0, networkClient.warmUpConnections(localPlainTextDataNodes, 100, TIME_FOR_WARM_UP_MS, responseInfos)); Assert.assertEquals("Mismatch in timeout responses", localPlainTextDataNodes.size() * MAX_PORTS_PLAIN_TEXT, responseInfos.size()); selector.setState(MockSelectorState.Good); }
@Override public int warmUpConnections(List<DataNodeId> dataNodeIds, int connectionWarmUpPercentagePerDataNode, long timeForWarmUp, List<ResponseInfo> responseInfoList) { logger.info("Connection warm up start."); if (dataNodeIds.size() == 0) { return 0; } dataNodeIds.forEach(dataNodeId -> connectionTracker.setMinimumActiveConnectionsPercentage(dataNodeId, connectionWarmUpPercentagePerDataNode)); int expectedConnections = connectionTracker.replenishConnections(this::connect, Integer.MAX_VALUE); long startTime = System.currentTimeMillis(); int successfulConnections = 0; int failedConnections = 0; while (successfulConnections + failedConnections < expectedConnections) { try { selector.poll(1000L); successfulConnections += selector.connected().size(); failedConnections += selector.disconnected().size(); handleSelectorEvents(responseInfoList); } catch (IOException e) { logger.error("Warm up received unexpected error while polling: ", e); } if (System.currentTimeMillis() - startTime > timeForWarmUp) { break; } } logger.info("Connection warm up done. Tried: {}, Succeeded: {}, Failed: {}, Time elapsed: {} ms", expectedConnections, successfulConnections, failedConnections, System.currentTimeMillis() - startTime); return successfulConnections; }
SocketNetworkClient implements NetworkClient { @Override public int warmUpConnections(List<DataNodeId> dataNodeIds, int connectionWarmUpPercentagePerDataNode, long timeForWarmUp, List<ResponseInfo> responseInfoList) { logger.info("Connection warm up start."); if (dataNodeIds.size() == 0) { return 0; } dataNodeIds.forEach(dataNodeId -> connectionTracker.setMinimumActiveConnectionsPercentage(dataNodeId, connectionWarmUpPercentagePerDataNode)); int expectedConnections = connectionTracker.replenishConnections(this::connect, Integer.MAX_VALUE); long startTime = System.currentTimeMillis(); int successfulConnections = 0; int failedConnections = 0; while (successfulConnections + failedConnections < expectedConnections) { try { selector.poll(1000L); successfulConnections += selector.connected().size(); failedConnections += selector.disconnected().size(); handleSelectorEvents(responseInfoList); } catch (IOException e) { logger.error("Warm up received unexpected error while polling: ", e); } if (System.currentTimeMillis() - startTime > timeForWarmUp) { break; } } logger.info("Connection warm up done. Tried: {}, Succeeded: {}, Failed: {}, Time elapsed: {} ms", expectedConnections, successfulConnections, failedConnections, System.currentTimeMillis() - startTime); return successfulConnections; } }
SocketNetworkClient implements NetworkClient { @Override public int warmUpConnections(List<DataNodeId> dataNodeIds, int connectionWarmUpPercentagePerDataNode, long timeForWarmUp, List<ResponseInfo> responseInfoList) { logger.info("Connection warm up start."); if (dataNodeIds.size() == 0) { return 0; } dataNodeIds.forEach(dataNodeId -> connectionTracker.setMinimumActiveConnectionsPercentage(dataNodeId, connectionWarmUpPercentagePerDataNode)); int expectedConnections = connectionTracker.replenishConnections(this::connect, Integer.MAX_VALUE); long startTime = System.currentTimeMillis(); int successfulConnections = 0; int failedConnections = 0; while (successfulConnections + failedConnections < expectedConnections) { try { selector.poll(1000L); successfulConnections += selector.connected().size(); failedConnections += selector.disconnected().size(); handleSelectorEvents(responseInfoList); } catch (IOException e) { logger.error("Warm up received unexpected error while polling: ", e); } if (System.currentTimeMillis() - startTime > timeForWarmUp) { break; } } logger.info("Connection warm up done. Tried: {}, Succeeded: {}, Failed: {}, Time elapsed: {} ms", expectedConnections, successfulConnections, failedConnections, System.currentTimeMillis() - startTime); return successfulConnections; } SocketNetworkClient(Selector selector, NetworkConfig networkConfig, NetworkMetrics networkMetrics,
      int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl, int checkoutTimeoutMs, Time time); }
SocketNetworkClient implements NetworkClient { @Override public int warmUpConnections(List<DataNodeId> dataNodeIds, int connectionWarmUpPercentagePerDataNode, long timeForWarmUp, List<ResponseInfo> responseInfoList) { logger.info("Connection warm up start."); if (dataNodeIds.size() == 0) { return 0; } dataNodeIds.forEach(dataNodeId -> connectionTracker.setMinimumActiveConnectionsPercentage(dataNodeId, connectionWarmUpPercentagePerDataNode)); int expectedConnections = connectionTracker.replenishConnections(this::connect, Integer.MAX_VALUE); long startTime = System.currentTimeMillis(); int successfulConnections = 0; int failedConnections = 0; while (successfulConnections + failedConnections < expectedConnections) { try { selector.poll(1000L); successfulConnections += selector.connected().size(); failedConnections += selector.disconnected().size(); handleSelectorEvents(responseInfoList); } catch (IOException e) { logger.error("Warm up received unexpected error while polling: ", e); } if (System.currentTimeMillis() - startTime > timeForWarmUp) { break; } } logger.info("Connection warm up done. Tried: {}, Succeeded: {}, Failed: {}, Time elapsed: {} ms", expectedConnections, successfulConnections, failedConnections, System.currentTimeMillis() - startTime); return successfulConnections; } SocketNetworkClient(Selector selector, NetworkConfig networkConfig, NetworkMetrics networkMetrics,
      int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl, int checkoutTimeoutMs, Time time); @Override List<ResponseInfo> sendAndPoll(List<RequestInfo> requestsToSend, Set<Integer> requestsToDrop,
      int pollTimeoutMs); @Override int warmUpConnections(List<DataNodeId> dataNodeIds, int connectionWarmUpPercentagePerDataNode,
      long timeForWarmUp, List<ResponseInfo> responseInfoList); @Override void close(); @Override void wakeup(); }
SocketNetworkClient implements NetworkClient { @Override public int warmUpConnections(List<DataNodeId> dataNodeIds, int connectionWarmUpPercentagePerDataNode, long timeForWarmUp, List<ResponseInfo> responseInfoList) { logger.info("Connection warm up start."); if (dataNodeIds.size() == 0) { return 0; } dataNodeIds.forEach(dataNodeId -> connectionTracker.setMinimumActiveConnectionsPercentage(dataNodeId, connectionWarmUpPercentagePerDataNode)); int expectedConnections = connectionTracker.replenishConnections(this::connect, Integer.MAX_VALUE); long startTime = System.currentTimeMillis(); int successfulConnections = 0; int failedConnections = 0; while (successfulConnections + failedConnections < expectedConnections) { try { selector.poll(1000L); successfulConnections += selector.connected().size(); failedConnections += selector.disconnected().size(); handleSelectorEvents(responseInfoList); } catch (IOException e) { logger.error("Warm up received unexpected error while polling: ", e); } if (System.currentTimeMillis() - startTime > timeForWarmUp) { break; } } logger.info("Connection warm up done. Tried: {}, Succeeded: {}, Failed: {}, Time elapsed: {} ms", expectedConnections, successfulConnections, failedConnections, System.currentTimeMillis() - startTime); return successfulConnections; } SocketNetworkClient(Selector selector, NetworkConfig networkConfig, NetworkMetrics networkMetrics,
      int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl, int checkoutTimeoutMs, Time time); @Override List<ResponseInfo> sendAndPoll(List<RequestInfo> requestsToSend, Set<Integer> requestsToDrop,
      int pollTimeoutMs); @Override int warmUpConnections(List<DataNodeId> dataNodeIds, int connectionWarmUpPercentagePerDataNode,
      long timeForWarmUp, List<ResponseInfo> responseInfoList); @Override void close(); @Override void wakeup();  }