@Test
public void testRecoveredFailure() throws Exception {

  final ReadRowsResponse result = ReadRowsResponse.getDefaultInstance();
  final Status errorStatus = Status.UNAVAILABLE;
  final AtomicInteger counter = new AtomicInteger(0);
  Answer<Void> answer =
      new Answer<Void>() {
        @Override
        public Void answer(InvocationOnMock invocation) throws Throwable {
          Listener listener = invocation.getArgument(1);
          if (counter.incrementAndGet() < 5) {
            listener.onClose(errorStatus, null);
          } else {
            listener.onMessage(result);
            listener.onClose(Status.OK, null);
          }
          return null;
        }
      };
  doAnswer(answer)
      .when(readAsync)
      .start(
          Mockito.<ReadRowsRequest>any(),
          Mockito.<ClientCall.Listener>any(),
          Mockito.<Metadata>any(),
          Mockito.<ClientCall>any());
  ListenableFuture future = createOperation(CallOptions.DEFAULT).getAsyncResult();
  Assert.assertEquals(result, future.get(1, TimeUnit.SECONDS));
  Assert.assertEquals(5, counter.get());
}