/**
 * Table mapping exception type to retry policy used for the exception in
 * write and watch request.
 * ---------------------------------------------------------------------------
 * |        Exception            | RetryPolicy for     | RetryPolicy for     |
 * |                             | Write request       | Watch request       |
 * |-------------------------------------------------------------------------|
 * | NotReplicatedException      | NO_RETRY            | NO_RETRY            |
 * |-------------------------------------------------------------------------|
 * | GroupMismatchException      | NO_RETRY            | NO_RETRY            |
 * |-------------------------------------------------------------------------|
 * | StateMachineException       | NO_RETRY            | NO_RETRY            |
 * |-------------------------------------------------------------------------|
 * | TimeoutIOException          | EXPONENTIAL_BACKOFF | NO_RETRY            |
 * |-------------------------------------------------------------------------|
 * | ResourceUnavailableException| EXPONENTIAL_BACKOFF | EXPONENTIAL_BACKOFF |
 * |-------------------------------------------------------------------------|
 * | Others                      | MULTILINEAR_RANDOM  | MULTILINEAR_RANDOM  |
 * |                             | _RETRY             | _RETRY               |
 * ---------------------------------------------------------------------------
 */
public static RetryPolicy createRetryPolicy(ConfigurationSource conf) {
  RatisClientConfig ratisClientConfig = OzoneConfiguration.of(conf)
      .getObject(RatisClientConfig.class);
  ExponentialBackoffRetry exponentialBackoffRetry =
      createExponentialBackoffPolicy(ratisClientConfig);
  MultipleLinearRandomRetry multipleLinearRandomRetry =
      MultipleLinearRandomRetry
          .parseCommaSeparated(ratisClientConfig.getMultilinearPolicy());

  long writeTimeout = ratisClientConfig.getWriteRequestTimeoutInMs();
  long watchTimeout = ratisClientConfig.getWatchRequestTimeoutInMs();

  return RequestTypeDependentRetryPolicy.newBuilder()
      .setRetryPolicy(RaftProtos.RaftClientRequestProto.TypeCase.WRITE,
          createExceptionDependentPolicy(exponentialBackoffRetry,
              multipleLinearRandomRetry, exponentialBackoffRetry))
      .setRetryPolicy(RaftProtos.RaftClientRequestProto.TypeCase.WATCH,
          createExceptionDependentPolicy(exponentialBackoffRetry,
              multipleLinearRandomRetry, RetryPolicies.noRetry()))
      .setTimeout(RaftProtos.RaftClientRequestProto.TypeCase.WRITE,
          TimeDuration.valueOf(writeTimeout, TimeUnit.MILLISECONDS))
      .setTimeout(RaftProtos.RaftClientRequestProto.TypeCase.WATCH,
          TimeDuration.valueOf(watchTimeout, TimeUnit.MILLISECONDS))
      .build();
}
 

static void runTestWatchRequestClientTimeout(TestParameters p) throws Exception {
  final Logger LOG = p.log;

  CompletableFuture<RaftClientReply> watchReply;
  // watch 1000 which will never be committed
  // so client can not receive reply, and connection closed, throw TimeoutException.
  // We should not retry, because if retry, RaftClientImpl::handleIOException will random select a leader,
  // then sometimes throw NotLeaderException.
  watchReply = p.sendWatchRequest(1000, RetryPolicies.noRetry());

  try {
    watchReply.get();
    fail("runTestWatchRequestClientTimeout failed");
  } catch (Exception ex) {
    LOG.error("error occurred", ex);
    Assert.assertTrue(ex.getCause().getClass() == AlreadyClosedException.class ||
        ex.getCause().getClass() == RaftRetryFailureException.class);
    if (ex.getCause() != null) {
      if (ex.getCause().getCause() != null) {
        Assert.assertEquals(TimeoutIOException.class,
            ex.getCause().getCause().getClass());
      }
    }
  }
}
 

private static ExceptionDependentRetry createExceptionDependentPolicy(
    ExponentialBackoffRetry exponentialBackoffRetry,
    MultipleLinearRandomRetry multipleLinearRandomRetry,
    RetryPolicy timeoutPolicy) {
  ExceptionDependentRetry.Builder builder =
      ExceptionDependentRetry.newBuilder();
  for (Class c : NO_RETRY_EXCEPTIONS) {
    builder.setExceptionToPolicy(c, RetryPolicies.noRetry());
  }
  return builder.setExceptionToPolicy(ResourceUnavailableException.class,
          exponentialBackoffRetry)
      .setExceptionToPolicy(TimeoutIOException.class, timeoutPolicy)
      .setDefaultPolicy(multipleLinearRandomRetry)
      .build();
}
 

@Override
public Action handleAttemptFailure(Event event) {
  if (!(event instanceof ClientRetryEvent)) {
    return RetryPolicies.retryForeverNoSleep().handleAttemptFailure(event);
  }
  final ClientRetryEvent clientEvent = (ClientRetryEvent) event;
  final TimeDuration timeout = timeoutMap.get(clientEvent.getRequest().getType().getTypeCase());
  if (timeout != null && clientEvent.isRequestTimeout(timeout)) {
    return NO_RETRY_ACTION;
  }
  return Optional.ofNullable(
      retryPolicyMap.get(clientEvent.getRequest().getType().getTypeCase()))
      .orElse(RetryPolicies.retryForeverNoSleep())
      .handleAttemptFailure(event);
}
 

@Test
public void testRetryMultipleTimesWithFixedSleep() {
  RetryPolicy retryPolicy = RetryPolicies
      .retryUpToMaximumCountWithFixedSleep(2,
          TimeDuration.valueOf(1000L, TimeUnit.MILLISECONDS));
   boolean shouldRetry = retryPolicy.shouldRetry(1);
  Assert.assertTrue(shouldRetry);
  Assert.assertTrue(1000 == retryPolicy.getSleepTime().getDuration());
  Assert.assertFalse(retryPolicy.shouldRetry(3));
}
 

private RetryPolicy getDefaultRetryPolicy() {
  return RetryPolicies.retryForeverWithSleep(RETRY_INTERVAL_DEFAULT);
}
 

private RetryPolicy getDefaultRetryPolicy() {
  return RetryPolicies.retryForeverWithSleep(RETRY_INTERVAL_DEFAULT);
}