@Test
  public void testRouterWithPreProcessQuotaCallback() throws Exception {
    try {
      Properties properties = getNonBlockingRouterProperties("DC1");
      properties.setProperty(QuotaConfig.BANDWIDTH_THROTTLING_FEATURE_ENABLED, "true");
      properties.setProperty(QuotaConfig.REQUEST_THROTTLING_ENABLED, "true");
      properties.setProperty(QuotaConfig.REQUEST_QUOTA_ENFORCER_SOURCE_PAIR_INFO_JSON,
          buildQuotaEnforcerSourceInfoPair());
      properties.setProperty(QuotaConfig.THROTTLING_MODE, quotaMode.name());
      properties.setProperty(QuotaConfig.CU_QUOTA_AGGREGATION_WINDOW_IN_SECS, "86400");
      properties.setProperty(RouterConfig.OPERATION_CONTROLLER, QuotaAwareOperationController.class.getCanonicalName());
      setRouter(properties, mockServerLayout, new LoggingNotificationSystem());
      assertExpectedThreadCounts(2, 1);
      AtomicInteger listenerCalledCount = new AtomicInteger(0);

      QuotaConfig quotaConfig = new QuotaConfig(new VerifiableProperties(properties));
      Account account = accountService.createAndAddRandomAccount(QuotaResourceType.ACCOUNT);
      QuotaMetrics quotaMetrics = new QuotaMetrics(new MetricRegistry());
      ChargeTesterQuotaManager chargeTesterQuotaManager =
          new ChargeTesterQuotaManager(quotaConfig, new SimpleQuotaRecommendationMergePolicy(quotaConfig),
              accountService, null, quotaMetrics, listenerCalledCount);
      chargeTesterQuotaManager.init();
      TestCUQuotaSource quotaSource = chargeTesterQuotaManager.getTestCuQuotaSource();
      RestRequest restRequest =
          createRestRequest(RestMethod.POST.name(), null, account, account.getAllContainers().iterator().next());
      QuotaChargeCallback quotaChargeCallback =
          QuotaUtils.buildQuotaChargeCallback(restRequest, chargeTesterQuotaManager, true);
      int blobSize = 3000;
      setOperationParams(blobSize, TTL_SECS, account.getId(), account.getAllContainers().iterator().next().getId());
      quotaSource.updateNewQuotaResources(Collections.singletonList(account));
      quotaSource.getCuQuota().put(String.valueOf(account.getId()), new CapacityUnit(8, 8));
      quotaSource.getCuUsage().put(String.valueOf(account.getId()), new CapacityUnit());
      String compositeBlobId =
          router.putBlob(null, putBlobProperties, putUserMetadata, putChannel, PutBlobOptions.DEFAULT, null,
              quotaChargeCallback).get();
      CapacityUnit quotaUsage = quotaSource.getCuUsage().get(String.valueOf(account.getId()));
      Assert.assertEquals(8, quotaUsage.getWcu());
      Assert.assertEquals(0, quotaUsage.getRcu());
      RetainingAsyncWritableChannel retainingAsyncWritableChannel = new RetainingAsyncWritableChannel();
      restRequest =
          createRestRequest(RestMethod.GET.name(), null, account, account.getAllContainers().iterator().next());
      quotaChargeCallback = QuotaUtils.buildQuotaChargeCallback(restRequest, chargeTesterQuotaManager, true);
      router.getBlob(compositeBlobId, new GetBlobOptionsBuilder().build(), null, quotaChargeCallback)
          .get()
          .getBlobDataChannel()
          .readInto(retainingAsyncWritableChannel, null)
          .get();
      Assert.assertEquals(0, (int) routerMetrics.outOfQuotaResourcesInQueue.getValue());
      Assert.assertEquals(0, (int) routerMetrics.delayedQuotaResourcesInQueue.getValue());
      // read out all the chunks.
      retainingAsyncWritableChannel.consumeContentAsInputStream().close();
      quotaUsage = quotaSource.getCuUsage().get(String.valueOf(account.getId()));
      Assert.assertEquals(8, quotaUsage.getWcu());
      Assert.assertEquals(8, quotaUsage.getRcu());
      retainingAsyncWritableChannel = new RetainingAsyncWritableChannel();
      try {
        router.getBlob(compositeBlobId, new GetBlobOptionsBuilder().build(), null, quotaChargeCallback).get();
        if (quotaMode == QuotaMode.THROTTLING) {
          fail("getBlob should throw exception if request is be rejected by an enforcer and quota mode is throttling.");
        }
      } catch (ExecutionException ex) {
        if (quotaMode == QuotaMode.TRACKING) {
          fail(
              "getBlob should not throw exception if request is be rejected by an enforcer and quota mode is tracking.");
        } else {
          Assert.assertEquals(RouterErrorCode.TooManyRequests, ((RouterException) ex.getCause()).getErrorCode());
        }
      }
      Assert.assertTrue(
          quotaMetrics.perQuotaResourceOutOfQuotaMap.get(Integer.toString(account.getId())).getCount() > 0);
      Assert.assertTrue(
          quotaMetrics.perQuotaResourceWouldBeThrottledMap.get(Integer.toString(account.getId())).getCount() > 0);
      Assert.assertEquals(quotaMetrics.frontendUsageReadAmplificationGauge.getValue(),
          (Integer) ChargeTesterQuotaManager.routerConfig.routerGetRequestParallelism);
      Assert.assertEquals(quotaMetrics.frontendUsageWriteAmplificationGauge.getValue(),
          (Integer) ChargeTesterQuotaManager.routerConfig.routerPutRequestParallelism);
      retainingAsyncWritableChannel.consumeContentAsInputStream().close();
    } finally {
      if (router != null) {
        router.close();
        assertExpectedThreadCounts(0, 0);

        //submission after closing should return a future that is already done.
        assertClosed();
      }
    }
  }
Test for {@link com.github.ambry.quota.PreProcessQuotaChargeCallback}.
ambry-router/src/test/java/com/github/ambry/router/NonBlockingQuotaTest.java