@Override
  public Future<RestResponse> restRequest(RestRequest request)
  {
    return restRequest(request, new RequestContext());
  }
@Test(dataProvider = "isD2Async")
  public void testStatsConsumerUpdateAndRemove(boolean isD2Async) throws Exception
  {
    AtomicReference<ServiceProperties> serviceProperties = new AtomicReference<>();
    TestBackupRequestsStrategyStatsConsumer statsConsumer = new TestBackupRequestsStrategyStatsConsumer();
    serviceProperties.set(createServiceProperties(null));
    BackupRequestsClient client = createClient(serviceProperties::get, statsConsumer, isD2Async);
    URI uri = URI.create("d2://testService");
    RestRequest restRequest = new RestRequestBuilder(uri).setEntity(CONTENT).build();
    RequestContext requestContext = new RequestContext();
    requestContext.putLocalAttr(R2Constants.OPERATION, "get");
    Future<RestResponse> response = client.restRequest(restRequest, requestContext);
    assertEquals(response.get().getStatus(), 200);
    List<StatsConsumerEvent> events = statsConsumer.getEvents();
    assertEquals(events.size(), 0);

    serviceProperties
        .set(createServiceProperties(Arrays.asList(createBackupRequestsConfiguration(5, "get"),
            createBackupRequestsConfiguration(1, "batch_get"))));

    requestContext = new RequestContext();
    requestContext.putLocalAttr(R2Constants.OPERATION, "get");
    response = client.restRequest(restRequest, requestContext);
    assertEquals(response.get().getStatus(), 200);
    events = statsConsumer.getEvents();
    assertEquals(events.size(), 2);
    assertEquals(events.get(0).isEventAdd(), true);
    assertEquals(events.get(0).getService(), SERVICE_NAME);
    assertEquals(events.get(0).getOperation(), "get");
    BackupRequestsStrategyStatsProvider statsProvider1 = events.get(0).getStatsProvider();
    assertNotNull(statsProvider1);
    assertEquals(events.get(1).isEventAdd(), true);
    assertEquals(events.get(1).getService(), SERVICE_NAME);
    assertEquals(events.get(1).getOperation(), "batch_get");
    BackupRequestsStrategyStatsProvider statsProvider2 = events.get(1).getStatsProvider();
    assertNotNull(statsProvider2);

    serviceProperties
        .set(createServiceProperties(Arrays.asList(createBackupRequestsConfiguration(1, "get"))));
    requestContext = new RequestContext();
    requestContext.putLocalAttr(R2Constants.OPERATION, "get");
    response = client.restRequest(restRequest, requestContext);
    assertEquals(response.get().getStatus(), 200);
    events = statsConsumer.getEvents();
    assertEquals(events.size(), 5);
    assertEquals(events.get(2).isEventAdd(), false);
    assertEquals(events.get(2).getService(), SERVICE_NAME);
    assertEquals(events.get(2).getOperation(), "get");
    BackupRequestsStrategyStatsProvider removedStatsProvider = events.get(2).getStatsProvider();
    assertNotNull(removedStatsProvider);
    assertSame(statsProvider1, removedStatsProvider);
    assertEquals(events.get(3).isEventAdd(), true);
    assertEquals(events.get(3).getService(), SERVICE_NAME);
    assertEquals(events.get(3).getOperation(), "get");
    BackupRequestsStrategyStatsProvider statsProvider3 = events.get(3).getStatsProvider();
    assertNotNull(statsProvider1);
    assertNotSame(statsProvider1, statsProvider3);

    assertEquals(events.get(4).isEventAdd(), false);
    assertEquals(events.get(4).getService(), SERVICE_NAME);
    assertEquals(events.get(4).getOperation(), "batch_get");
    BackupRequestsStrategyStatsProvider removedStatsProvider2 = events.get(4).getStatsProvider();
    assertNotNull(removedStatsProvider);
    assertSame(statsProvider2, removedStatsProvider2);
  }