@Test
public void testScannerIdle() throws IOException {

  ReadRowsRequest.Builder requestBuilder = ReadRowsRequest.newBuilder().setTableName(TABLE_NAME);
  requestBuilder.getRowsBuilder().addRowKeys(ByteString.EMPTY);
  ResultScanner<FlatRow> scanner = defaultClient.readFlatRows(requestBuilder.build());
  ArgumentCaptor<ClientCall.Listener> listenerCaptor =
      ArgumentCaptor.forClass(ClientCall.Listener.class);
  verify(mockClientCall, times(1)).start(listenerCaptor.capture(), any(Metadata.class));
  Listener listener = listenerCaptor.getValue();
  ByteString key1 = ByteString.copyFromUtf8("Key1");
  ByteString key2 = ByteString.copyFromUtf8("Key2");
  listener.onMessage(RetryingReadRowsOperationTest.buildResponse(key1));
  listener.onMessage(RetryingReadRowsOperationTest.buildResponse(key2));
  listener.onClose(
      Status.CANCELLED.withCause(
          new Watchdog.StreamWaitTimeoutException(
              Watchdog.State.IDLE, TimeUnit.MINUTES.toMillis(10))),
      new Metadata());
  Assert.assertEquals(2, scanner.available());
  Assert.assertEquals(key1, scanner.next().getRowKey());
  Assert.assertEquals(key2, scanner.next().getRowKey());
  ByteString key3 = ByteString.copyFromUtf8("Key3");
  listener.onMessage(RetryingReadRowsOperationTest.buildResponse(key3));
  Assert.assertEquals(key3, scanner.next().getRowKey());
  // There was a retry based on the idle
  verify(mochScheduler, times(1)).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
}