@Test
public void testThrottled() throws Exception {

  final SettableFuture registerInvoked = SettableFuture.create();
  final SettableFuture registerWaiter = SettableFuture.create();
  final long id = 1l;
  when(mockResourceLimiter.registerOperationWithHeapSize(anyLong()))
      .then(
          new Answer<Long>() {
            @Override
            public Long answer(InvocationOnMock invocation) throws Throwable {
              registerInvoked.set("");
              registerWaiter.get(1, TimeUnit.SECONDS);
              return id;
            }
          });
  final ThrottlingClientInterceptor underTest =
      new ThrottlingClientInterceptor(mockResourceLimiter);
  final int numMessages = 5;
  Future<?> future =
      executorService.submit(
          new Runnable() {
            @Override
            public void run() {
              ClientCall call =
                  underTest.interceptCall(methodDescriptor, CallOptions.DEFAULT, mockChannel);
              call.start(mockListener, new Metadata());
              call.request(numMessages);
              // This should block
              call.sendMessage(request);
            }
          });
  registerInvoked.get(1, TimeUnit.SECONDS);
  verify(mockResourceLimiter, times(1))
      .registerOperationWithHeapSize(eq((long) request.getSerializedSize()));
  verify(mockChannel, times(0)).newCall(any(MethodDescriptor.class), any(CallOptions.class));
  registerWaiter.set("");
  future.get(1, TimeUnit.SECONDS);
  verify(mockChannel, times(1)).newCall(any(MethodDescriptor.class), any(CallOptions.class));
  verify(mockClientCall, times(1)).request(eq(numMessages));
  ArgumentCaptor<ClientCall.Listener> listenerArgumentCaptor =
      ArgumentCaptor.forClass(ClientCall.Listener.class);
  verify(mockClientCall, times(1)).start(listenerArgumentCaptor.capture(), any(Metadata.class));
  Status status = Status.OK;
  Metadata trailers = new Metadata();
  listenerArgumentCaptor.getValue().onClose(status, trailers);
  verify(mockResourceLimiter, times(1)).markCanBeCompleted(eq(id));
  verify(mockListener, times(1)).onClose(same(status), same(trailers));
}