@Test public void testResponseWithNullRequestInfo() throws Exception { Properties props = getNonBlockingRouterProperties("DC1"); VerifiableProperties verifiableProperties = new VerifiableProperties((props)); RouterConfig routerConfig = new RouterConfig(verifiableProperties); routerMetrics = new NonBlockingRouterMetrics(mockClusterMap, routerConfig); NetworkClient mockNetworkClient = Mockito.mock(NetworkClient.class); Mockito.when(mockNetworkClient.warmUpConnections(anyList(), anyInt(), anyLong(), anyList())).thenReturn(1); doNothing().when(mockNetworkClient).close(); List<ResponseInfo> responseInfoList = new ArrayList<>(); MockDataNodeId testDataNode = (MockDataNodeId) mockClusterMap.getDataNodeIds().get(0); responseInfoList.add(new ResponseInfo(null, NetworkClientErrorCode.NetworkError, null, testDataNode)); CountDownLatch invocationLatch = new CountDownLatch(3); doAnswer(invocation -> { invocationLatch.countDown(); return responseInfoList; }).when(mockNetworkClient).sendAndPoll(anyList(), anySet(), anyInt()); NetworkClientFactory networkClientFactory = Mockito.mock(NetworkClientFactory.class); Mockito.when(networkClientFactory.getNetworkClient()).thenReturn(mockNetworkClient); NonBlockingRouter testRouter = new NonBlockingRouter(routerConfig, routerMetrics, networkClientFactory, new LoggingNotificationSystem(), mockClusterMap, kms, cryptoService, cryptoJobHandler, accountService, mockTime, MockClusterMap.DEFAULT_PARTITION_CLASS); assertTrue("Invocation latch didn't count to 0 within 10 seconds", invocationLatch.await(10, TimeUnit.SECONDS)); assertTrue("The node should be considered timeout", testDataNode.isTimedOut()); testRouter.close(); }
@Override public void close() { shutDownOperationControllers(); waitForResponseHandlerThreadExit(); if (kms != null) { try { kms.close(); } catch (IOException e) { logger.error("Exception thrown on closing KMS", e); } } if (cryptoJobHandler != null) { cryptoJobHandler.close(); } for (Closeable resource : resourcesToClose) { try { resource.close(); } catch (IOException e) { logger.error("Exception thrown on closing {}", resource.getClass().getName()); } } routerMetrics.close(); }
NonBlockingRouter implements Router { @Override public void close() { shutDownOperationControllers(); waitForResponseHandlerThreadExit(); if (kms != null) { try { kms.close(); } catch (IOException e) { logger.error("Exception thrown on closing KMS", e); } } if (cryptoJobHandler != null) { cryptoJobHandler.close(); } for (Closeable resource : resourcesToClose) { try { resource.close(); } catch (IOException e) { logger.error("Exception thrown on closing {}", resource.getClass().getName()); } } routerMetrics.close(); } }
NonBlockingRouter implements Router { @Override public void close() { shutDownOperationControllers(); waitForResponseHandlerThreadExit(); if (kms != null) { try { kms.close(); } catch (IOException e) { logger.error("Exception thrown on closing KMS", e); } } if (cryptoJobHandler != null) { cryptoJobHandler.close(); } for (Closeable resource : resourcesToClose) { try { resource.close(); } catch (IOException e) { logger.error("Exception thrown on closing {}", resource.getClass().getName()); } } routerMetrics.close(); } NonBlockingRouter(RouterConfig routerConfig, NonBlockingRouterMetrics routerMetrics,
      NetworkClientFactory networkClientFactory, NotificationSystem notificationSystem, ClusterMap clusterMap,
      KeyManagementService kms, CryptoService cryptoService, CryptoJobHandler cryptoJobHandler,
      AccountService accountService, Time time, String defaultPartitionClass); }
NonBlockingRouter implements Router { @Override public void close() { shutDownOperationControllers(); waitForResponseHandlerThreadExit(); if (kms != null) { try { kms.close(); } catch (IOException e) { logger.error("Exception thrown on closing KMS", e); } } if (cryptoJobHandler != null) { cryptoJobHandler.close(); } for (Closeable resource : resourcesToClose) { try { resource.close(); } catch (IOException e) { logger.error("Exception thrown on closing {}", resource.getClass().getName()); } } routerMetrics.close(); } NonBlockingRouter(RouterConfig routerConfig, NonBlockingRouterMetrics routerMetrics,
      NetworkClientFactory networkClientFactory, NotificationSystem notificationSystem, ClusterMap clusterMap,
      KeyManagementService kms, CryptoService cryptoService, CryptoJobHandler cryptoJobHandler,
      AccountService accountService, Time time, String defaultPartitionClass); @Override Future<GetBlobResult> getBlob(String blobIdStr, GetBlobOptions options,
      final Callback<GetBlobResult> callback); @Override Future<String> putBlob(BlobProperties blobProperties, byte[] userMetadata, ReadableStreamChannel channel,
      PutBlobOptions options, Callback<String> callback); @Override Future<String> stitchBlob(BlobProperties blobProperties, byte[] userMetadata, List<ChunkInfo> chunksToStitch,
      Callback<String> callback); @Override Future<Void> deleteBlob(String blobId, String serviceId, Callback<Void> callback); @Override Future<Void> undeleteBlob(String blobId, String serviceId, Callback<Void> callback); @Override Future<Void> updateBlobTtl(String blobId, String serviceId, long expiresAtMs, Callback<Void> callback); @Override void close(); }
NonBlockingRouter implements Router { @Override public void close() { shutDownOperationControllers(); waitForResponseHandlerThreadExit(); if (kms != null) { try { kms.close(); } catch (IOException e) { logger.error("Exception thrown on closing KMS", e); } } if (cryptoJobHandler != null) { cryptoJobHandler.close(); } for (Closeable resource : resourcesToClose) { try { resource.close(); } catch (IOException e) { logger.error("Exception thrown on closing {}", resource.getClass().getName()); } } routerMetrics.close(); } NonBlockingRouter(RouterConfig routerConfig, NonBlockingRouterMetrics routerMetrics,
      NetworkClientFactory networkClientFactory, NotificationSystem notificationSystem, ClusterMap clusterMap,
      KeyManagementService kms, CryptoService cryptoService, CryptoJobHandler cryptoJobHandler,
      AccountService accountService, Time time, String defaultPartitionClass); @Override Future<GetBlobResult> getBlob(String blobIdStr, GetBlobOptions options,
      final Callback<GetBlobResult> callback); @Override Future<String> putBlob(BlobProperties blobProperties, byte[] userMetadata, ReadableStreamChannel channel,
      PutBlobOptions options, Callback<String> callback); @Override Future<String> stitchBlob(BlobProperties blobProperties, byte[] userMetadata, List<ChunkInfo> chunksToStitch,
      Callback<String> callback); @Override Future<Void> deleteBlob(String blobId, String serviceId, Callback<Void> callback); @Override Future<Void> undeleteBlob(String blobId, String serviceId, Callback<Void> callback); @Override Future<Void> updateBlobTtl(String blobId, String serviceId, long expiresAtMs, Callback<Void> callback); @Override void close();  }