@Test public void testClose() { List<RequestInfo> requestInfoList = new ArrayList<RequestInfo>(); networkClient.close(); try { networkClient.sendAndPoll(requestInfoList, Collections.emptySet(), POLL_TIMEOUT_MS); Assert.fail("Polling after close should throw"); } catch (IllegalStateException e) { } }
@Override public void close() { Set<RequestMetadata> requestMetadataToRelease = new HashSet<>(); if (pendingRequests != null) { requestMetadataToRelease.addAll(pendingRequests); } if (connectionIdToRequestInFlight != null) { requestMetadataToRelease.addAll(connectionIdToRequestInFlight.values()); } if (pendingConnectionsToAssociatedRequests != null) { requestMetadataToRelease.addAll(pendingConnectionsToAssociatedRequests.values()); } requestMetadataToRelease.forEach(requestMetadata -> { if (!requestMetadata.requestInfo.getRequest().isSendComplete()) { requestMetadata.requestInfo.getRequest().release(); } }); logger.trace("Closing the SocketNetworkClient"); selector.close(); closed = true; }
SocketNetworkClient implements NetworkClient { @Override public void close() { Set<RequestMetadata> requestMetadataToRelease = new HashSet<>(); if (pendingRequests != null) { requestMetadataToRelease.addAll(pendingRequests); } if (connectionIdToRequestInFlight != null) { requestMetadataToRelease.addAll(connectionIdToRequestInFlight.values()); } if (pendingConnectionsToAssociatedRequests != null) { requestMetadataToRelease.addAll(pendingConnectionsToAssociatedRequests.values()); } requestMetadataToRelease.forEach(requestMetadata -> { if (!requestMetadata.requestInfo.getRequest().isSendComplete()) { requestMetadata.requestInfo.getRequest().release(); } }); logger.trace("Closing the SocketNetworkClient"); selector.close(); closed = true; } }
SocketNetworkClient implements NetworkClient { @Override public void close() { Set<RequestMetadata> requestMetadataToRelease = new HashSet<>(); if (pendingRequests != null) { requestMetadataToRelease.addAll(pendingRequests); } if (connectionIdToRequestInFlight != null) { requestMetadataToRelease.addAll(connectionIdToRequestInFlight.values()); } if (pendingConnectionsToAssociatedRequests != null) { requestMetadataToRelease.addAll(pendingConnectionsToAssociatedRequests.values()); } requestMetadataToRelease.forEach(requestMetadata -> { if (!requestMetadata.requestInfo.getRequest().isSendComplete()) { requestMetadata.requestInfo.getRequest().release(); } }); logger.trace("Closing the SocketNetworkClient"); selector.close(); closed = true; } SocketNetworkClient(Selector selector, NetworkConfig networkConfig, NetworkMetrics networkMetrics,
      int maxConnectionsPerPortPlainText, int maxConnectionsPerPortSsl, int checkoutTimeoutMs, Time time); }
SocketNetworkClient implements NetworkClient { @Override public void close() { Set<RequestMetadata> requestMetadataToRelease = new HashSet<>(); if (pendingRequests != null) { requestMetadataToRelease.addAll(pendingRequests); } if (connectionIdToRequestInFlight != null) { requestMetadataToRelease.addAll(connectionIdToRequestInFlight.values()); } if (pendingConnectionsToAssociatedRequests != null) { requestMetadataToRelease.addAll(pendingConnectionsToAssociatedRequests.values()); } requestMetadataToRelease.forEach(requestMetadata -> { if (!requestMetadata.requestInfo.getRequest().isSendComplete()) { requestMetadata.requestInfo.getRequest().release(); } }); logger.trace("Closing the SocketNetworkClient"); selector.close(); closed = true; } 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 void close() { Set<RequestMetadata> requestMetadataToRelease = new HashSet<>(); if (pendingRequests != null) { requestMetadataToRelease.addAll(pendingRequests); } if (connectionIdToRequestInFlight != null) { requestMetadataToRelease.addAll(connectionIdToRequestInFlight.values()); } if (pendingConnectionsToAssociatedRequests != null) { requestMetadataToRelease.addAll(pendingConnectionsToAssociatedRequests.values()); } requestMetadataToRelease.forEach(requestMetadata -> { if (!requestMetadata.requestInfo.getRequest().isSendComplete()) { requestMetadata.requestInfo.getRequest().release(); } }); logger.trace("Closing the SocketNetworkClient"); selector.close(); closed = true; } 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();  }