import com.github.rholder.retry.Attempt; //导入依赖的package包/类
/**
 * Build Retryer.
 * - If Writer implements Retriable, it will use the RetryerBuilder from the writer.
 * - Otherwise, it will use DEFAULT writer builder.
 *
 * - If Gobblin metrics is enabled, it will emit all failure count in to metrics.
 *
 * @param state
 * @return
 */
private Retryer<Void> buildRetryer(State state) {
  RetryerBuilder<Void> builder = null;
  if (writer instanceof Retriable) {
    builder = ((Retriable) writer).getRetryerBuilder();
  } else {
    builder = createRetryBuilder(state);
  }

  if (GobblinMetrics.isEnabled(state)) {
    final Optional<Meter> retryMeter = Optional.of(Instrumented.getMetricContext(state, getClass()).meter(FAILED_RETRY_WRITES_METER));

    builder.withRetryListener(new RetryListener() {
      @Override
      public <V> void onRetry(Attempt<V> attempt) {
        if (attempt.hasException()) {
          LOG.warn("Caught exception. This may be retried.", attempt.getExceptionCause());
          Instrumented.markMeter(retryMeter);
          failedWrites++;
        }
      }
    });
  }
  return builder.build();
}
 

import com.github.rholder.retry.Attempt; //导入依赖的package包/类
private Retryer<Response> buildResponseRetryer() {
	final RetryListener retryListener = new RetryListener() {
		@Override
		public <V> void onRetry(Attempt<V> attempt) {
			final long attemptNumber = attempt.getAttemptNumber();
			if (attemptNumber > ONE) {
				final long delaySinceFirstAttemptInMilliseconds = attempt.getDelaySinceFirstAttempt();
				final long delaySinceFirstAttemptInSeconds = TimeUnit.SECONDS
						.convert(delaySinceFirstAttemptInMilliseconds, TimeUnit.MILLISECONDS);
				final Response response = (Response) attempt.getResult();
				response.bufferEntity();
				LOGGER.warn(String.format(RETRY_ATTEMPT_MESSAGE, delaySinceFirstAttemptInSeconds, attemptNumber,
						response.getStatus(), response.readEntity(String.class)));
			}
		}
	};

	final long maximumWaitDuration = requestRetryConfiguration.getMininumWaitDuration() * TWO;
	return RetryerBuilder.<Response> newBuilder().retryIfResult(this::shouldRetryResponse)
			.withWaitStrategy(WaitStrategies.randomWait(requestRetryConfiguration.getMininumWaitDuration(),
					requestRetryConfiguration.getMininumWaitUnit(), maximumWaitDuration,
					requestRetryConfiguration.getMininumWaitUnit()))
			.withRetryListener(retryListener)
			.withStopStrategy(StopStrategies.stopAfterDelay(requestRetryConfiguration.getTimeoutDuration(),
					requestRetryConfiguration.getTimeoutUnit()))
			.build();
}
 

import com.github.rholder.retry.Attempt; //导入依赖的package包/类
CommandQueries(List<QueryRequest> queries, Keyspace keyspace) {
    super(Setter
            .withGroupKey(HystrixCommandGroupKey.Factory.asKey("BatchExecutor"))
            .andThreadPoolPropertiesDefaults(
                    HystrixThreadPoolProperties.Setter()
                            .withCoreSize(threadPoolCoreSize)
                            // Sizing these two based on the thread pool core size
                            .withQueueSizeRejectionThreshold(
                                    threadPoolCoreSize * QUEUE_MULTIPLIER)
                            .withMaxQueueSize(threadPoolCoreSize * QUEUE_MULTIPLIER))
            .andCommandPropertiesDefaults(
                    HystrixCommandProperties.Setter()
                            .withExecutionTimeoutEnabled(false)
                            .withExecutionTimeoutInMilliseconds(timeoutMs)
                            .withRequestLogEnabled(requestLogEnabled)));

    this.queries = queries;
    this.keyspace = keyspace;
    this.graqlExecuteTimer = metricRegistry.timer(name(this.getClass(), "execute"));
    this.attemptMeter = metricRegistry.meter(name(this.getClass(), "attempt"));
    this.retryer = RetryerBuilder.<List<QueryResponse>>newBuilder()
            .retryIfException((throwable) ->
                    throwable instanceof GraknClientException
                            && ((GraknClientException) throwable).isRetriable())
            .retryIfExceptionOfType(ConnectException.class)
            .withWaitStrategy(WaitStrategies.exponentialWait(10, 1, TimeUnit.MINUTES))
            .withStopStrategy(StopStrategies.stopAfterAttempt(maxRetries + 1))
            .withRetryListener(new RetryListener() {
                @Override
                public <V> void onRetry(Attempt<V> attempt) {
                    attemptMeter.mark();
                }
            })
            .build();
}
 

import com.github.rholder.retry.Attempt; //导入依赖的package包/类
@Override
public <V> void onRetry(Attempt<V> attempt) {
  attemptCount = attempt.getAttemptNumber();
  if (delegate != null) {
    delegate.onRetry(attempt);
  }
}
 

import com.github.rholder.retry.Attempt; //导入依赖的package包/类
public GuaranteedDeliveryTransport( long maxWaitTimeSeconds, int maxAttempts ) {
   retryer = RetryerBuilder.<Boolean>newBuilder()
      .withRetryListener( new RetryListener() {
         @Override
         public <V> void onRetry( Attempt<V> attempt ) {
            String errorMessage = attempt.hasException() ? attempt.getExceptionCause().toString() : "no errors";
            V result = attempt.hasResult() ? attempt.getResult() : null;
            log.info( "Attempt: {},  result: {}, error: {}", attempt.getAttemptNumber(), result, errorMessage );
         }
      } )
      .retryIfException( e -> !( e instanceof InterruptedException ) )
      .withWaitStrategy( WaitStrategies.fibonacciWait( maxWaitTimeSeconds, TimeUnit.SECONDS ) )
      .withStopStrategy( StopStrategies.stopAfterAttempt( maxAttempts ) )
      .build();
}
 

import com.github.rholder.retry.Attempt; //导入依赖的package包/类
private <V> void onRestorePollPodsAttempt(Attempt<V> attempt) {
  if (attempt.hasException()) {
    LOG.warn("restore: failed polling pods, attempt = {}", attempt.getAttemptNumber(), attempt.getExceptionCause());
  }
}
 

import com.github.rholder.retry.Attempt; //导入依赖的package包/类
@Override
public <V> void onRetry(Attempt<V> attempt) {
    if (attempt.hasException()) {
        LOG.warn("Retrying transaction after {" + attempt.getAttemptNumber() + "} attempts due to exception {" + attempt.getExceptionCause().getMessage() + "}");
    }
}
 

import com.github.rholder.retry.Attempt; //导入依赖的package包/类
@Override
public <V> void onRetry(Attempt<V> attempt) {
  lastAttemptNumber = attempt.getAttemptNumber();
}