@Test public void testRequestInfoNullInResponse() throws Exception { NetworkClient compositeClient = factory.getNetworkClient(); when(diskMocks.client.sendAndPoll(any(), any(), anyInt())).thenReturn( Collections.singletonList(new ResponseInfo(null, NetworkClientErrorCode.NetworkError, null))); compositeClient.sendAndPoll(Collections.emptyList(), Collections.emptySet(), 1000); verify(diskMocks.client).sendAndPoll(Collections.emptyList(), Collections.emptySet(), 1000); }
@Override public List<ResponseInfo> sendAndPoll(List<RequestInfo> allRequestsToSend, Set<Integer> allRequestsToDrop, int pollTimeoutMs) { EnumMap<ReplicaType, List<RequestInfo>> requestsToSendByType = new EnumMap<>(ReplicaType.class); EnumMap<ReplicaType, Set<Integer>> requestsToDropByType = new EnumMap<>(ReplicaType.class); for (ReplicaType replicaType : childNetworkClients.keySet()) { requestsToSendByType.put(replicaType, new ArrayList<>()); requestsToDropByType.put(replicaType, new HashSet<>()); } for (RequestInfo requestInfo : allRequestsToSend) { ReplicaType replicaType = requestInfo.getReplicaId().getReplicaType(); List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType); if (requestsToSend == null) { throw new IllegalStateException("No NetworkClient configured for replica type: " + replicaType); } requestsToSend.add(requestInfo); correlationIdToReplicaType.put(requestInfo.getRequest().getCorrelationId(), replicaType); } for (Integer correlationId : allRequestsToDrop) { ReplicaType replicaType = correlationIdToReplicaType.get(correlationId); if (replicaType != null) { requestsToDropByType.get(replicaType).add(correlationId); } } AtomicBoolean wakeupCalled = new AtomicBoolean(false); ArrayList<Future<List<ResponseInfo>>> sendAndPollFutures = new ArrayList<>(childNetworkClients.size()); childNetworkClients.forEach((replicaType, client) -> { List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType); Set<Integer> requestsToDrop = requestsToDropByType.get(replicaType); if (!requestsToSend.isEmpty() || !requestsToDrop.isEmpty()) { logger.trace("replicaType={}, requestsToSend={}, requestsToDrop={}", replicaType, requestsToSend, requestsToDrop); } sendAndPollFutures.add(executor.submit(() -> { List<ResponseInfo> childClientResponses = client.sendAndPoll(requestsToSend, requestsToDrop, pollTimeoutMs); if (wakeupCalled.compareAndSet(false, true)) { childNetworkClients.values().stream().filter(c -> c != client).forEach(NetworkClient::wakeup); } return childClientResponses; })); }); List<ResponseInfo> responses = new ArrayList<>(); for (Future<List<ResponseInfo>> future : sendAndPollFutures) { try { List<ResponseInfo> responseInfoList = future.get(); for (ResponseInfo responseInfo : responseInfoList) { if (responseInfo.getRequestInfo() != null) { correlationIdToReplicaType.remove(responseInfo.getRequestInfo().getRequest().getCorrelationId()); } responses.add(responseInfo); } } catch (InterruptedException | ExecutionException e) { logger.error("Hit unexpected exception on parallel sendAndPoll.", e); } } return responses; }
CompositeNetworkClient implements NetworkClient { @Override public List<ResponseInfo> sendAndPoll(List<RequestInfo> allRequestsToSend, Set<Integer> allRequestsToDrop, int pollTimeoutMs) { EnumMap<ReplicaType, List<RequestInfo>> requestsToSendByType = new EnumMap<>(ReplicaType.class); EnumMap<ReplicaType, Set<Integer>> requestsToDropByType = new EnumMap<>(ReplicaType.class); for (ReplicaType replicaType : childNetworkClients.keySet()) { requestsToSendByType.put(replicaType, new ArrayList<>()); requestsToDropByType.put(replicaType, new HashSet<>()); } for (RequestInfo requestInfo : allRequestsToSend) { ReplicaType replicaType = requestInfo.getReplicaId().getReplicaType(); List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType); if (requestsToSend == null) { throw new IllegalStateException("No NetworkClient configured for replica type: " + replicaType); } requestsToSend.add(requestInfo); correlationIdToReplicaType.put(requestInfo.getRequest().getCorrelationId(), replicaType); } for (Integer correlationId : allRequestsToDrop) { ReplicaType replicaType = correlationIdToReplicaType.get(correlationId); if (replicaType != null) { requestsToDropByType.get(replicaType).add(correlationId); } } AtomicBoolean wakeupCalled = new AtomicBoolean(false); ArrayList<Future<List<ResponseInfo>>> sendAndPollFutures = new ArrayList<>(childNetworkClients.size()); childNetworkClients.forEach((replicaType, client) -> { List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType); Set<Integer> requestsToDrop = requestsToDropByType.get(replicaType); if (!requestsToSend.isEmpty() || !requestsToDrop.isEmpty()) { logger.trace("replicaType={}, requestsToSend={}, requestsToDrop={}", replicaType, requestsToSend, requestsToDrop); } sendAndPollFutures.add(executor.submit(() -> { List<ResponseInfo> childClientResponses = client.sendAndPoll(requestsToSend, requestsToDrop, pollTimeoutMs); if (wakeupCalled.compareAndSet(false, true)) { childNetworkClients.values().stream().filter(c -> c != client).forEach(NetworkClient::wakeup); } return childClientResponses; })); }); List<ResponseInfo> responses = new ArrayList<>(); for (Future<List<ResponseInfo>> future : sendAndPollFutures) { try { List<ResponseInfo> responseInfoList = future.get(); for (ResponseInfo responseInfo : responseInfoList) { if (responseInfo.getRequestInfo() != null) { correlationIdToReplicaType.remove(responseInfo.getRequestInfo().getRequest().getCorrelationId()); } responses.add(responseInfo); } } catch (InterruptedException | ExecutionException e) { logger.error("Hit unexpected exception on parallel sendAndPoll.", e); } } return responses; } }
CompositeNetworkClient implements NetworkClient { @Override public List<ResponseInfo> sendAndPoll(List<RequestInfo> allRequestsToSend, Set<Integer> allRequestsToDrop, int pollTimeoutMs) { EnumMap<ReplicaType, List<RequestInfo>> requestsToSendByType = new EnumMap<>(ReplicaType.class); EnumMap<ReplicaType, Set<Integer>> requestsToDropByType = new EnumMap<>(ReplicaType.class); for (ReplicaType replicaType : childNetworkClients.keySet()) { requestsToSendByType.put(replicaType, new ArrayList<>()); requestsToDropByType.put(replicaType, new HashSet<>()); } for (RequestInfo requestInfo : allRequestsToSend) { ReplicaType replicaType = requestInfo.getReplicaId().getReplicaType(); List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType); if (requestsToSend == null) { throw new IllegalStateException("No NetworkClient configured for replica type: " + replicaType); } requestsToSend.add(requestInfo); correlationIdToReplicaType.put(requestInfo.getRequest().getCorrelationId(), replicaType); } for (Integer correlationId : allRequestsToDrop) { ReplicaType replicaType = correlationIdToReplicaType.get(correlationId); if (replicaType != null) { requestsToDropByType.get(replicaType).add(correlationId); } } AtomicBoolean wakeupCalled = new AtomicBoolean(false); ArrayList<Future<List<ResponseInfo>>> sendAndPollFutures = new ArrayList<>(childNetworkClients.size()); childNetworkClients.forEach((replicaType, client) -> { List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType); Set<Integer> requestsToDrop = requestsToDropByType.get(replicaType); if (!requestsToSend.isEmpty() || !requestsToDrop.isEmpty()) { logger.trace("replicaType={}, requestsToSend={}, requestsToDrop={}", replicaType, requestsToSend, requestsToDrop); } sendAndPollFutures.add(executor.submit(() -> { List<ResponseInfo> childClientResponses = client.sendAndPoll(requestsToSend, requestsToDrop, pollTimeoutMs); if (wakeupCalled.compareAndSet(false, true)) { childNetworkClients.values().stream().filter(c -> c != client).forEach(NetworkClient::wakeup); } return childClientResponses; })); }); List<ResponseInfo> responses = new ArrayList<>(); for (Future<List<ResponseInfo>> future : sendAndPollFutures) { try { List<ResponseInfo> responseInfoList = future.get(); for (ResponseInfo responseInfo : responseInfoList) { if (responseInfo.getRequestInfo() != null) { correlationIdToReplicaType.remove(responseInfo.getRequestInfo().getRequest().getCorrelationId()); } responses.add(responseInfo); } } catch (InterruptedException | ExecutionException e) { logger.error("Hit unexpected exception on parallel sendAndPoll.", e); } } return responses; } CompositeNetworkClient(EnumMap<ReplicaType, NetworkClient> childNetworkClients); }
CompositeNetworkClient implements NetworkClient { @Override public List<ResponseInfo> sendAndPoll(List<RequestInfo> allRequestsToSend, Set<Integer> allRequestsToDrop, int pollTimeoutMs) { EnumMap<ReplicaType, List<RequestInfo>> requestsToSendByType = new EnumMap<>(ReplicaType.class); EnumMap<ReplicaType, Set<Integer>> requestsToDropByType = new EnumMap<>(ReplicaType.class); for (ReplicaType replicaType : childNetworkClients.keySet()) { requestsToSendByType.put(replicaType, new ArrayList<>()); requestsToDropByType.put(replicaType, new HashSet<>()); } for (RequestInfo requestInfo : allRequestsToSend) { ReplicaType replicaType = requestInfo.getReplicaId().getReplicaType(); List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType); if (requestsToSend == null) { throw new IllegalStateException("No NetworkClient configured for replica type: " + replicaType); } requestsToSend.add(requestInfo); correlationIdToReplicaType.put(requestInfo.getRequest().getCorrelationId(), replicaType); } for (Integer correlationId : allRequestsToDrop) { ReplicaType replicaType = correlationIdToReplicaType.get(correlationId); if (replicaType != null) { requestsToDropByType.get(replicaType).add(correlationId); } } AtomicBoolean wakeupCalled = new AtomicBoolean(false); ArrayList<Future<List<ResponseInfo>>> sendAndPollFutures = new ArrayList<>(childNetworkClients.size()); childNetworkClients.forEach((replicaType, client) -> { List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType); Set<Integer> requestsToDrop = requestsToDropByType.get(replicaType); if (!requestsToSend.isEmpty() || !requestsToDrop.isEmpty()) { logger.trace("replicaType={}, requestsToSend={}, requestsToDrop={}", replicaType, requestsToSend, requestsToDrop); } sendAndPollFutures.add(executor.submit(() -> { List<ResponseInfo> childClientResponses = client.sendAndPoll(requestsToSend, requestsToDrop, pollTimeoutMs); if (wakeupCalled.compareAndSet(false, true)) { childNetworkClients.values().stream().filter(c -> c != client).forEach(NetworkClient::wakeup); } return childClientResponses; })); }); List<ResponseInfo> responses = new ArrayList<>(); for (Future<List<ResponseInfo>> future : sendAndPollFutures) { try { List<ResponseInfo> responseInfoList = future.get(); for (ResponseInfo responseInfo : responseInfoList) { if (responseInfo.getRequestInfo() != null) { correlationIdToReplicaType.remove(responseInfo.getRequestInfo().getRequest().getCorrelationId()); } responses.add(responseInfo); } } catch (InterruptedException | ExecutionException e) { logger.error("Hit unexpected exception on parallel sendAndPoll.", e); } } return responses; } CompositeNetworkClient(EnumMap<ReplicaType, NetworkClient> childNetworkClients); @Override List<ResponseInfo> sendAndPoll(List<RequestInfo> allRequestsToSend, Set<Integer> allRequestsToDrop,
      int pollTimeoutMs); @Override int warmUpConnections(List<DataNodeId> dataNodeIds, int connectionWarmUpPercentagePerDataNode,
      long timeForWarmUp, List<ResponseInfo> responseInfoList); @Override void wakeup(); @Override void close(); }
CompositeNetworkClient implements NetworkClient { @Override public List<ResponseInfo> sendAndPoll(List<RequestInfo> allRequestsToSend, Set<Integer> allRequestsToDrop, int pollTimeoutMs) { EnumMap<ReplicaType, List<RequestInfo>> requestsToSendByType = new EnumMap<>(ReplicaType.class); EnumMap<ReplicaType, Set<Integer>> requestsToDropByType = new EnumMap<>(ReplicaType.class); for (ReplicaType replicaType : childNetworkClients.keySet()) { requestsToSendByType.put(replicaType, new ArrayList<>()); requestsToDropByType.put(replicaType, new HashSet<>()); } for (RequestInfo requestInfo : allRequestsToSend) { ReplicaType replicaType = requestInfo.getReplicaId().getReplicaType(); List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType); if (requestsToSend == null) { throw new IllegalStateException("No NetworkClient configured for replica type: " + replicaType); } requestsToSend.add(requestInfo); correlationIdToReplicaType.put(requestInfo.getRequest().getCorrelationId(), replicaType); } for (Integer correlationId : allRequestsToDrop) { ReplicaType replicaType = correlationIdToReplicaType.get(correlationId); if (replicaType != null) { requestsToDropByType.get(replicaType).add(correlationId); } } AtomicBoolean wakeupCalled = new AtomicBoolean(false); ArrayList<Future<List<ResponseInfo>>> sendAndPollFutures = new ArrayList<>(childNetworkClients.size()); childNetworkClients.forEach((replicaType, client) -> { List<RequestInfo> requestsToSend = requestsToSendByType.get(replicaType); Set<Integer> requestsToDrop = requestsToDropByType.get(replicaType); if (!requestsToSend.isEmpty() || !requestsToDrop.isEmpty()) { logger.trace("replicaType={}, requestsToSend={}, requestsToDrop={}", replicaType, requestsToSend, requestsToDrop); } sendAndPollFutures.add(executor.submit(() -> { List<ResponseInfo> childClientResponses = client.sendAndPoll(requestsToSend, requestsToDrop, pollTimeoutMs); if (wakeupCalled.compareAndSet(false, true)) { childNetworkClients.values().stream().filter(c -> c != client).forEach(NetworkClient::wakeup); } return childClientResponses; })); }); List<ResponseInfo> responses = new ArrayList<>(); for (Future<List<ResponseInfo>> future : sendAndPollFutures) { try { List<ResponseInfo> responseInfoList = future.get(); for (ResponseInfo responseInfo : responseInfoList) { if (responseInfo.getRequestInfo() != null) { correlationIdToReplicaType.remove(responseInfo.getRequestInfo().getRequest().getCorrelationId()); } responses.add(responseInfo); } } catch (InterruptedException | ExecutionException e) { logger.error("Hit unexpected exception on parallel sendAndPoll.", e); } } return responses; } CompositeNetworkClient(EnumMap<ReplicaType, NetworkClient> childNetworkClients); @Override List<ResponseInfo> sendAndPoll(List<RequestInfo> allRequestsToSend, Set<Integer> allRequestsToDrop,
      int pollTimeoutMs); @Override int warmUpConnections(List<DataNodeId> dataNodeIds, int connectionWarmUpPercentagePerDataNode,
      long timeForWarmUp, List<ResponseInfo> responseInfoList); @Override void wakeup(); @Override void close();  }