@Test public void testExceptionOnPoll() { List<RequestInfo> requestInfoList = new ArrayList<>(); requestInfoList.add(new RequestInfo(sslHost, sslPort, new MockSend(3), replicaOnSslNode)); requestInfoList.add(new RequestInfo(sslHost, sslPort, new MockSend(4), replicaOnSslNode)); selector.setState(MockSelectorState.ThrowExceptionOnPoll); try { List<ResponseInfo> responseInfoList = networkClient.sendAndPoll(requestInfoList, Collections.emptySet(), POLL_TIMEOUT_MS); responseInfoList.forEach(ResponseInfo::release); } catch (Exception e) { Assert.fail("If selector throws on poll, sendAndPoll() should not throw."); } requestInfoList.forEach(requestInfo -> requestInfo.getRequest().release()); selector.setState(MockSelectorState.Good); }
@Override public List<ResponseInfo> sendAndPoll(List<RequestInfo> requestsToSend, Set<Integer> requestsToDrop, int pollTimeoutMs) { if (closed || !selector.isOpen()) { throw new IllegalStateException("The SocketNetworkClient is closed."); } long startTime = time.milliseconds(); List<ResponseInfo> responseInfoList = new ArrayList<>(); List<NetworkSend> sends = null; try { for (RequestInfo requestInfo : requestsToSend) { pendingRequests.add(new RequestMetadata(requestInfo)); } sends = prepareSends(requestsToDrop, responseInfoList); replenishConnections(); for (Integer correlationId : requestsToDrop) { String connectionId = correlationIdInFlightToConnectionId.get(correlationId); if (connectionId != null) { selector.close(connectionId); } } selector.poll(pollTimeoutMs, sends); handleSelectorEvents(responseInfoList); } catch (Exception e) { logger.error("Received an unexpected error during sendAndPoll(): ", e); networkMetrics.networkClientException.inc(); } finally { numPendingRequests.set(pendingRequests.size()); networkMetrics.networkClientSendAndPollTime.update(time.milliseconds() - startTime, TimeUnit.MILLISECONDS); } return responseInfoList; }
SocketNetworkClient implements NetworkClient { @Override public List<ResponseInfo> sendAndPoll(List<RequestInfo> requestsToSend, Set<Integer> requestsToDrop, int pollTimeoutMs) { if (closed || !selector.isOpen()) { throw new IllegalStateException("The SocketNetworkClient is closed."); } long startTime = time.milliseconds(); List<ResponseInfo> responseInfoList = new ArrayList<>(); List<NetworkSend> sends = null; try { for (RequestInfo requestInfo : requestsToSend) { pendingRequests.add(new RequestMetadata(requestInfo)); } sends = prepareSends(requestsToDrop, responseInfoList); replenishConnections(); for (Integer correlationId : requestsToDrop) { String connectionId = correlationIdInFlightToConnectionId.get(correlationId); if (connectionId != null) { selector.close(connectionId); } } selector.poll(pollTimeoutMs, sends); handleSelectorEvents(responseInfoList); } catch (Exception e) { logger.error("Received an unexpected error during sendAndPoll(): ", e); networkMetrics.networkClientException.inc(); } finally { numPendingRequests.set(pendingRequests.size()); networkMetrics.networkClientSendAndPollTime.update(time.milliseconds() - startTime, TimeUnit.MILLISECONDS); } return responseInfoList; } }
SocketNetworkClient implements NetworkClient { @Override public List<ResponseInfo> sendAndPoll(List<RequestInfo> requestsToSend, Set<Integer> requestsToDrop, int pollTimeoutMs) { if (closed || !selector.isOpen()) { throw new IllegalStateException("The SocketNetworkClient is closed."); } long startTime = time.milliseconds(); List<ResponseInfo> responseInfoList = new ArrayList<>(); List<NetworkSend> sends = null; try { for (RequestInfo requestInfo : requestsToSend) { pendingRequests.add(new RequestMetadata(requestInfo)); } sends = prepareSends(requestsToDrop, responseInfoList); replenishConnections(); for (Integer correlationId : requestsToDrop) { String connectionId = correlationIdInFlightToConnectionId.get(correlationId); if (connectionId != null) { selector.close(connectionId); } } selector.poll(pollTimeoutMs, sends); handleSelectorEvents(responseInfoList); } catch (Exception e) { logger.error("Received an unexpected error during sendAndPoll(): ", e); networkMetrics.networkClientException.inc(); } finally { numPendingRequests.set(pendingRequests.size()); networkMetrics.networkClientSendAndPollTime.update(time.milliseconds() - startTime, TimeUnit.MILLISECONDS); } return responseInfoList; } SocketNetworkClient(Selector selector, NetworkConfig networkConfig, NetworkMetrics networkMetrics,
      int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl, int checkoutTimeoutMs, Time time); }
SocketNetworkClient implements NetworkClient { @Override public List<ResponseInfo> sendAndPoll(List<RequestInfo> requestsToSend, Set<Integer> requestsToDrop, int pollTimeoutMs) { if (closed || !selector.isOpen()) { throw new IllegalStateException("The SocketNetworkClient is closed."); } long startTime = time.milliseconds(); List<ResponseInfo> responseInfoList = new ArrayList<>(); List<NetworkSend> sends = null; try { for (RequestInfo requestInfo : requestsToSend) { pendingRequests.add(new RequestMetadata(requestInfo)); } sends = prepareSends(requestsToDrop, responseInfoList); replenishConnections(); for (Integer correlationId : requestsToDrop) { String connectionId = correlationIdInFlightToConnectionId.get(correlationId); if (connectionId != null) { selector.close(connectionId); } } selector.poll(pollTimeoutMs, sends); handleSelectorEvents(responseInfoList); } catch (Exception e) { logger.error("Received an unexpected error during sendAndPoll(): ", e); networkMetrics.networkClientException.inc(); } finally { numPendingRequests.set(pendingRequests.size()); networkMetrics.networkClientSendAndPollTime.update(time.milliseconds() - startTime, TimeUnit.MILLISECONDS); } return responseInfoList; } 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 List<ResponseInfo> sendAndPoll(List<RequestInfo> requestsToSend, Set<Integer> requestsToDrop, int pollTimeoutMs) { if (closed || !selector.isOpen()) { throw new IllegalStateException("The SocketNetworkClient is closed."); } long startTime = time.milliseconds(); List<ResponseInfo> responseInfoList = new ArrayList<>(); List<NetworkSend> sends = null; try { for (RequestInfo requestInfo : requestsToSend) { pendingRequests.add(new RequestMetadata(requestInfo)); } sends = prepareSends(requestsToDrop, responseInfoList); replenishConnections(); for (Integer correlationId : requestsToDrop) { String connectionId = correlationIdInFlightToConnectionId.get(correlationId); if (connectionId != null) { selector.close(connectionId); } } selector.poll(pollTimeoutMs, sends); handleSelectorEvents(responseInfoList); } catch (Exception e) { logger.error("Received an unexpected error during sendAndPoll(): ", e); networkMetrics.networkClientException.inc(); } finally { numPendingRequests.set(pendingRequests.size()); networkMetrics.networkClientSendAndPollTime.update(time.milliseconds() - startTime, TimeUnit.MILLISECONDS); } return responseInfoList; } 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();  }