@VisibleForTesting
  protected <T> T retryOnException(ObjectStoreOperation<T> op,
                                   Supplier<String> description) throws IOException {
    RetryPolicy retryPolicy = getRetryPolicy();
    IOException thrownException = null;
    while (retryPolicy.attempt()) {
      try {
        return op.apply();
      } catch (IOException e) {
        LOG.debug("Attempt {} to {} failed with exception : {}", retryPolicy.getAttemptCount(),
            description.get(), e.toString());
        handleRetriablException(e);
        thrownException = e;
      }
    }
    throw thrownException;
  }
@Test
  public void testRetryOnException() {
    //if ufs throws SocketException, it will retry.
    ObjectUnderFileSystem.ObjectStoreOperation<Object> mockOp = Mockito.mock(
        ObjectUnderFileSystem.ObjectStoreOperation.class);
    try {
      Mockito.when(mockOp.apply()).thenThrow(new SocketException()).thenReturn(null);
      mObjectUFS.retryOnException(mockOp, () -> "test");
      // retry policy will retry, so op will apply twice.
      Mockito.verify(mockOp, Mockito.atLeast(2)).apply();
    } catch (IOException e) {
      fail();
    }

    //if ufs throws other Exception, it will throw Exception.
    try {
      Mockito.reset(mockOp);
      Mockito.when(mockOp.apply()).thenThrow(new FileNotFoundException()).thenReturn(null);
      mObjectUFS.retryOnException(mockOp, () -> "test");
    } catch (IOException e) {
      assertTrue(e instanceof FileNotFoundException);
    }
    try {
      // op only apply once and then throw exception.
      Mockito.verify(mockOp, Mockito.atLeast(1)).apply();
    } catch (IOException e) {
      fail();
    }
  }