@Test public void testBadCallback() throws Exception { RequestAndResult req = new RequestAndResult(chunkSize * 5 + random.nextInt(chunkSize - 1) + 1); router = getNonBlockingRouter(); final CountDownLatch callbackCalled = new CountDownLatch(1); requestAndResultsList.clear(); for (int i = 0; i < 4; i++) { requestAndResultsList.add(new RequestAndResult(chunkSize + random.nextInt(5) * random.nextInt(chunkSize))); } instantiateNewRouterForPuts = false; ReadableStreamChannel putChannel = new ByteBufferReadableStreamChannel(ByteBuffer.wrap(req.putContent)); Future future = router.putBlob(req.putBlobProperties, req.putUserMetadata, putChannel, req.options, (result, exception) -> { callbackCalled.countDown(); throw new RuntimeException("Throwing an exception in the user callback"); }); submitPutsAndAssertSuccess(false); future.get(); Assert.assertTrue("Callback not called.", callbackCalled.await(MAX_WAIT_MS, TimeUnit.MILLISECONDS)); assertEquals("All operations should be finished.", 0, router.getOperationsCount()); Assert.assertTrue("Router should not be closed", router.isOpen()); requestAndResultsList.clear(); requestAndResultsList.add(new RequestAndResult(chunkSize + random.nextInt(5) * random.nextInt(chunkSize))); instantiateNewRouterForPuts = false; submitPutsAndAssertSuccess(true); }
boolean isOpen() { return isOpen.get(); }
PutManager { boolean isOpen() { return isOpen.get(); } }
PutManager { boolean isOpen() { return isOpen.get(); } PutManager(ClusterMap clusterMap, ResponseHandler responseHandler, NotificationSystem notificationSystem,
      RouterConfig routerConfig, NonBlockingRouterMetrics routerMetrics, RouterCallback routerCallback, String suffix,
      KeyManagementService kms, CryptoService cryptoService, CryptoJobHandler cryptoJobHandler,
      AccountService accountService, Time time, String defaultPartitionClass); }
PutManager { boolean isOpen() { return isOpen.get(); } PutManager(ClusterMap clusterMap, ResponseHandler responseHandler, NotificationSystem notificationSystem,
      RouterConfig routerConfig, NonBlockingRouterMetrics routerMetrics, RouterCallback routerCallback, String suffix,
      KeyManagementService kms, CryptoService cryptoService, CryptoJobHandler cryptoJobHandler,
      AccountService accountService, Time time, String defaultPartitionClass);  }
PutManager { boolean isOpen() { return isOpen.get(); } PutManager(ClusterMap clusterMap, ResponseHandler responseHandler, NotificationSystem notificationSystem,
      RouterConfig routerConfig, NonBlockingRouterMetrics routerMetrics, RouterCallback routerCallback, String suffix,
      KeyManagementService kms, CryptoService cryptoService, CryptoJobHandler cryptoJobHandler,
      AccountService accountService, Time time, String defaultPartitionClass);   }