@Test public void testRequestDropping() { AtomicInteger nextCorrelationId = new AtomicInteger(1); Function<Integer, List<RequestInfo>> requestGen = numRequests -> IntStream.range(0, numRequests) .mapToObj( i -> new RequestInfo(sslHost, sslPort, new MockSend(nextCorrelationId.getAndIncrement()), replicaOnSslNode)) .collect(Collectors.toList()); List<ResponseInfo> responseInfoList; responseInfoList = networkClient.sendAndPoll(requestGen.apply(3), Collections.emptySet(), POLL_TIMEOUT_MS); Assert.assertEquals("No responses expected in first poll.", 0, responseInfoList.size()); responseInfoList = networkClient.sendAndPoll(Collections.emptyList(), new HashSet<>(Arrays.asList(2, 3)), POLL_TIMEOUT_MS); Assert.assertEquals("Should receive only as many responses as there were requests", 3, responseInfoList.size()); for (ResponseInfo responseInfo : responseInfoList) { MockSend send = (MockSend) responseInfo.getRequestInfo().getRequest(); if (send.getCorrelationId() == 1) { NetworkClientErrorCode error = responseInfo.getError(); ByteBuf response = responseInfo.content(); Assert.assertNull("Should not have encountered an error", error); Assert.assertNotNull("Should receive a valid response", response); int correlationIdInRequest = send.getCorrelationId(); int correlationIdInResponse = response.readInt(); Assert.assertEquals("Received response for the wrong request", correlationIdInRequest, correlationIdInResponse); } else { Assert.assertEquals("Expected connection unavailable on dropped request", NetworkClientErrorCode.ConnectionUnavailable, responseInfo.getError()); Assert.assertNull("Should not receive a response", responseInfo.content()); } } responseInfoList.forEach(ResponseInfo::release); selector.setState(MockSelectorState.IdlePoll); responseInfoList = networkClient.sendAndPoll(requestGen.apply(3), Collections.emptySet(), POLL_TIMEOUT_MS); Assert.assertEquals("No responses expected in idle poll.", 0, responseInfoList.size()); selector.setState(MockSelectorState.Good); responseInfoList = networkClient.sendAndPoll(Collections.emptyList(), Collections.singleton(4), POLL_TIMEOUT_MS); Assert.assertEquals("Should receive only as many responses as there were requests", 3, responseInfoList.size()); for (ResponseInfo responseInfo : responseInfoList) { MockSend send = (MockSend) responseInfo.getRequestInfo().getRequest(); if (send.getCorrelationId() != 4) { NetworkClientErrorCode error = responseInfo.getError(); ByteBuf response = responseInfo.content(); Assert.assertNull("Should not have encountered an error", error); Assert.assertNotNull("Should receive a valid response", response); int correlationIdInRequest = send.getCorrelationId(); int correlationIdInResponse = response.readInt(); Assert.assertEquals("Received response for the wrong request", correlationIdInRequest, correlationIdInResponse); responseInfo.release(); } else { Assert.assertEquals("Expected network error (from closed connection for dropped request)", NetworkClientErrorCode.NetworkError, responseInfo.getError()); Assert.assertNull("Should not receive a response", responseInfo.content()); responseInfo.release(); } } responseInfoList = networkClient.sendAndPoll(Collections.emptyList(), Collections.singleton(1), POLL_TIMEOUT_MS); Assert.assertEquals("No more responses expected.", 0, responseInfoList.size()); responseInfoList.forEach(ResponseInfo::release); }
@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();  }