Java 类com.amazonaws.retry.RetryPolicy 实例源码

/**
 * The client execution timer uses interrupts to abort the client but if another thread
 * interrupts the current thread for another reason we don't want to squash the
 * {@link InterruptedException}. We should set the thread's interrupted status and throw the
 * exception back out (we can't throw the actual {@link InterruptedException} because it's
 * checked)
 */
@Test(timeout = TEST_TIMEOUT)
public void interruptCausedBySomethingOtherThanTimer_PropagatesInterruptToCaller() {
    final int socketTimeoutInMillis = 100;
    httpClient = new AmazonHttpClient(new ClientConfiguration().withSocketTimeout(socketTimeoutInMillis)
            .withClientExecutionTimeout(CLIENT_EXECUTION_TIMEOUT)
            .withRetryPolicy(new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION,
                    new FixedTimeBackoffStrategy(CLIENT_EXECUTION_TIMEOUT), 1, false)));

    // We make sure the first connection has failed due to the socket timeout before
    // interrupting so we know that we are sleeping per the backoff strategy. Apache HTTP
    // client doesn't seem to honor interrupts reliably but Thread.sleep does
    interruptCurrentThreadAfterDelay(socketTimeoutInMillis * 2);

    try {
        httpClient.requestExecutionBuilder().request(newGetRequest()).execute();
        fail("Exception expected");
    } catch (AmazonClientException e) {
        assertTrue(Thread.currentThread().isInterrupted());
        assertThat(e.getCause(), instanceOf(InterruptedException.class));
    }
}

@Test(timeout = TEST_TIMEOUT)
public void clientExecutionTimeoutEnabled_WithShorterRequestTimeoutAndRetry_ThrowsClientExecutionTimeoutException()
        throws IOException {
    final int clientExecutionTimeout = 1500;
    final int requestTimeout = 1000;
    final int backoffTime = 300;
    httpClient = new AmazonHttpClient(new ClientConfiguration().withClientExecutionTimeout(clientExecutionTimeout)
            .withRequestTimeout(requestTimeout)
            .withRetryPolicy(new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION,
                    new FixedTimeBackoffStrategy(backoffTime), Integer.MAX_VALUE, false)));

    try {
        httpClient.requestExecutionBuilder().request(newGetRequest()).execute();
        fail("Exception expected");
    } catch (AmazonClientException e) {
        assertThat(e, instanceOf(ClientExecutionTimeoutException.class));
        // Completed tasks means the client execution was aborted by the timer
        assertNumberOfTasksTriggered(httpClient.getClientExecutionTimer(), 1);
        assertNumberOfTasksTriggered(httpClient.getHttpRequestTimer(), 1);
    }
}

private boolean shouldRetryCompleteMultipartUpload(AmazonWebServiceRequest originalRequest,
                                                   AmazonS3Exception exception,
                                                   int retriesAttempted) {

    final RetryPolicy retryPolicy = clientConfiguration.getRetryPolicy();

    if (retryPolicy == null || retryPolicy.getRetryCondition() == null) {
        return false;
    }

    if (retryPolicy == PredefinedRetryPolicies.NO_RETRY_POLICY) {
        return false;
    }

    return completeMultipartUploadRetryCondition.shouldRetry
            (originalRequest, exception, retriesAttempted);
}

public static AmazonS3Client getS3Client(AuthCredentialsServiceState credentials,
        String regionId) {

    ClientConfiguration configuration = new ClientConfiguration();
    configuration.withRetryPolicy(new RetryPolicy(new CustomRetryCondition(),
            DEFAULT_BACKOFF_STRATEGY,
            DEFAULT_MAX_ERROR_RETRY,
            false));

    AWSStaticCredentialsProvider awsStaticCredentialsProvider = new AWSStaticCredentialsProvider(
            new BasicAWSCredentials(credentials.privateKeyId,
                    EncryptionUtils.decrypt(credentials.privateKey)));

    AmazonS3ClientBuilder amazonS3ClientBuilder = AmazonS3ClientBuilder
            .standard()
            .withClientConfiguration(configuration)
            .withCredentials(awsStaticCredentialsProvider)
            .withRegion(regionId);

    if (isAwsClientMock()) {
        throw new IllegalArgumentException("AWS Mock does not support S3 client");
    }

    return (AmazonS3Client) amazonS3ClientBuilder.build();
}

public static AmazonEC2 getEC2SynchronousClient(AuthCredentialsServiceState credentials,
        String region) {
    ClientConfiguration configuration = new ClientConfiguration();
    configuration.withRetryPolicy(new RetryPolicy(new CustomRetryCondition(),
            DEFAULT_BACKOFF_STRATEGY,
            DEFAULT_MAX_ERROR_RETRY,
            true));

    AWSStaticCredentialsProvider awsStaticCredentialsProvider = new AWSStaticCredentialsProvider(
            new BasicAWSCredentials(credentials.privateKeyId,
                    EncryptionUtils.decrypt(credentials.privateKey)));

    AmazonEC2ClientBuilder ec2ClientBuilder = AmazonEC2ClientBuilder.standard()
            .withCredentials(awsStaticCredentialsProvider)
            .withRegion(region)
            .withClientConfiguration(configuration);

    return ec2ClientBuilder.build();
}

public AmazonS3 newInstance(Configuration conf) {
  int maxErrorRetry = conf.getInt(ConfigurationVariable.UPLOAD_RETRY_COUNT.getName(),
      ConfigurationVariable.UPLOAD_RETRY_COUNT.defaultIntValue());
  long errorRetryDelay = conf.getLong(ConfigurationVariable.UPLOAD_RETRY_DELAY_MS.getName(),
      ConfigurationVariable.UPLOAD_RETRY_DELAY_MS.defaultLongValue());

  LOG.info("Creating AWS S3 client with a retry policy of {} retries and {} ms of exponential backoff delay",
      maxErrorRetry, errorRetryDelay);

  RetryPolicy retryPolicy = new RetryPolicy(new CounterBasedRetryCondition(maxErrorRetry),
      new ExponentialBackoffStrategy(errorRetryDelay), maxErrorRetry, true);
  ClientConfiguration clientConfiguration = new ClientConfiguration();
  clientConfiguration.setRetryPolicy(retryPolicy);
  clientConfiguration.setMaxErrorRetry(maxErrorRetry);

  AmazonS3ClientBuilder builder = AmazonS3ClientBuilder
      .standard()
      .withCredentials(new HadoopAWSCredentialProviderChain(conf))
      .withClientConfiguration(clientConfiguration);

  EndpointConfiguration endpointConfiguration = getEndpointConfiguration(conf);
  if (endpointConfiguration != null) {
    builder.withEndpointConfiguration(endpointConfiguration);
  } else {
    builder.withRegion(getRegion(conf));
  }

  return builder.build();
}

public S3SQSClient(Region region, String queueName, String accessKey, String secretKey) {
    ClientConfiguration clientConfiguration = new ClientConfiguration();

    clientConfiguration.setRequestTimeout(5000);
    clientConfiguration.setRetryPolicy(new RetryPolicy(null, null, 3, true));

    if (accessKey.isEmpty() && secretKey.isEmpty()) {
        this.sqs = new AmazonSQSClient(new InstanceProfileCredentialsProvider(), clientConfiguration);
    } else {
        this.sqs = new AmazonSQSClient(new BasicAWSCredentials(accessKey, secretKey), clientConfiguration);
    }
    this.sqs.setRegion(region);

    this.queueName = queueName;
}

private ClientConfiguration getClientConfiguration() {
    ClientConfiguration clientConfig = new ClientConfiguration()
            .withUserAgentPrefix("CodeBuild-Jenkins-Plugin") //tags all calls made from Jenkins plugin.
            .withProxyHost(proxyHost)
            .withRetryPolicy(new RetryPolicy(new CodeBuildClientRetryCondition(),
                    new PredefinedBackoffStrategies.ExponentialBackoffStrategy(5000, 20000),
                    10, true));

    if(proxyPort != null) {
        clientConfig.setProxyPort(proxyPort);
    }
    return clientConfig;
}

public static AmazonEC2AsyncClient getAsyncClient(
        AuthCredentialsServiceState credentials, String region,
        ExecutorService executorService) {

    ClientConfiguration configuration = new ClientConfiguration();
    configuration.setMaxConnections(100);
    configuration.withRetryPolicy(new RetryPolicy(new CustomRetryCondition(),
            DEFAULT_BACKOFF_STRATEGY,
            DEFAULT_MAX_ERROR_RETRY,
            false));

    AWSStaticCredentialsProvider awsStaticCredentialsProvider = new AWSStaticCredentialsProvider(
            new BasicAWSCredentials(credentials.privateKeyId,
                    EncryptionUtils.decrypt(credentials.privateKey)));

    AmazonEC2AsyncClientBuilder ec2AsyncClientBuilder = AmazonEC2AsyncClientBuilder
            .standard()
            .withClientConfiguration(configuration)
            .withCredentials(awsStaticCredentialsProvider)
            .withExecutorFactory(() -> executorService);

    if (region == null) {
        region = Regions.DEFAULT_REGION.getName();
    }

    if (isAwsClientMock()) {
        configuration.addHeader(AWS_REGION_HEADER, region);
        ec2AsyncClientBuilder.setClientConfiguration(configuration);
        AwsClientBuilder.EndpointConfiguration endpointConfiguration = new AwsClientBuilder.EndpointConfiguration(
                getAWSMockHost() + AWS_MOCK_EC2_ENDPOINT, region);
        ec2AsyncClientBuilder.setEndpointConfiguration(endpointConfiguration);
    } else {
        ec2AsyncClientBuilder.setRegion(region);
    }

    return (AmazonEC2AsyncClient) ec2AsyncClientBuilder.build();
}

public static AmazonCloudWatchAsyncClient getStatsAsyncClient(
        AuthCredentialsServiceState credentials, String region,
        ExecutorService executorService, boolean isMockRequest) {

    ClientConfiguration configuration = new ClientConfiguration();
    configuration.withRetryPolicy(new RetryPolicy(new CustomRetryCondition(),
            DEFAULT_BACKOFF_STRATEGY,
            DEFAULT_MAX_ERROR_RETRY,
            false));

    AWSStaticCredentialsProvider awsStaticCredentialsProvider = new AWSStaticCredentialsProvider(
            new BasicAWSCredentials(credentials.privateKeyId,
                    EncryptionUtils.decrypt(credentials.privateKey)));

    AmazonCloudWatchAsyncClientBuilder amazonCloudWatchAsyncClientBuilder = AmazonCloudWatchAsyncClientBuilder
            .standard()
            .withClientConfiguration(configuration)
            .withCredentials(awsStaticCredentialsProvider)
            .withExecutorFactory(() -> executorService);

    if (region == null) {
        region = Regions.DEFAULT_REGION.getName();
    }

    if (isAwsClientMock()) {
        configuration.addHeader(AWS_REGION_HEADER, region);
        amazonCloudWatchAsyncClientBuilder.setClientConfiguration(configuration);
        AwsClientBuilder.EndpointConfiguration endpointConfiguration = new AwsClientBuilder.EndpointConfiguration(
                getAWSMockHost() + AWS_MOCK_CLOUDWATCH_ENDPOINT, region);
        amazonCloudWatchAsyncClientBuilder.setEndpointConfiguration(endpointConfiguration);
    } else {
        amazonCloudWatchAsyncClientBuilder.setRegion(region);
    }

    return (AmazonCloudWatchAsyncClient) amazonCloudWatchAsyncClientBuilder.build();
}

public static AmazonElasticLoadBalancingAsyncClient getLoadBalancingAsyncClient(
        AuthCredentialsServiceState credentials, String region,
        ExecutorService executorService) {

    ClientConfiguration configuration = new ClientConfiguration();
    configuration.withRetryPolicy(new RetryPolicy(new CustomRetryCondition(),
            DEFAULT_BACKOFF_STRATEGY,
            DEFAULT_MAX_ERROR_RETRY,
            false));

    AWSStaticCredentialsProvider awsStaticCredentialsProvider = new AWSStaticCredentialsProvider(
            new BasicAWSCredentials(credentials.privateKeyId,
                    EncryptionUtils.decrypt(credentials.privateKey)));

    AmazonElasticLoadBalancingAsyncClientBuilder amazonElasticLoadBalancingAsyncClientBuilder = AmazonElasticLoadBalancingAsyncClientBuilder
            .standard()
            .withClientConfiguration(configuration)
            .withCredentials(awsStaticCredentialsProvider)
            .withExecutorFactory(() -> executorService);

    if (region == null) {
        region = Regions.DEFAULT_REGION.getName();
    }

    if (isAwsClientMock()) {
        AwsClientBuilder.EndpointConfiguration endpointConfiguration = new AwsClientBuilder.EndpointConfiguration(
                getAWSMockHost() + AWS_MOCK_LOAD_BALANCING_ENDPOINT, region);
        amazonElasticLoadBalancingAsyncClientBuilder
                .setEndpointConfiguration(endpointConfiguration);
    } else {
        amazonElasticLoadBalancingAsyncClientBuilder.setRegion(region);
    }

    return (AmazonElasticLoadBalancingAsyncClient) amazonElasticLoadBalancingAsyncClientBuilder
            .build();
}

/**
 * Returns whether this method was able to initiate recovery from the given exception.
 *
 * <p>If recovery is possible and if the retry policy returned by {@link ClientConfiguration#getRetryPolicy()}
 * suggests to perform another transmission attempt, this methods schedules this task to be run again, with a
 * delay as returned by the back-off strategy of the retry policy.
 *
 * @param amazonClientException the exception that will be passed to the retry policy and the back-off strategy
 * @param getObjectRequest the S3 get-object request
 * @return whether this method was able to initiate recovery
 *
 * @see RetryPolicy.BackoffStrategy#delayBeforeNextRetry(com.amazonaws.AmazonWebServiceRequest, AmazonClientException, int)
 */
private boolean couldRecoverFromException(AmazonClientException amazonClientException,
    GetObjectRequest getObjectRequest) {

    RetryPolicy retryPolicy = s3ClientConfiguration.getRetryPolicy();
    int configuredMaxErrorRetry = s3ClientConfiguration.getMaxErrorRetry();
    // The following is copied from AmazonHttpClient#shouldRetry(): "We should use the maxErrorRetry in the
    // RetryPolicy if either the user has not explicitly set it in ClientConfiguration, or the RetryPolicy
    // is configured to take higher precedence."
    int maxErrorRetry = (configuredMaxErrorRetry < 0 || !retryPolicy.isMaxErrorRetryInClientConfigHonored())
        ? retryPolicy.getMaxErrorRetry()
        : configuredMaxErrorRetry;

    RetryPolicy.RetryCondition retryCondition = retryPolicy.getRetryCondition();
    boolean returnValue = false;
    if (retriesAttempted <= maxErrorRetry
        && retryCondition.shouldRetry(getObjectRequest, amazonClientException, retriesAttempted)) {

        long delay = retryPolicy.getBackoffStrategy()
            .delayBeforeNextRetry(getObjectRequest, amazonClientException, retriesAttempted);
        ++retriesAttempted;

        // java.util.concurrent states: "Actions in a thread prior to the submission of a Runnable to an
        // Executor happen-before its execution begins." Consequently, memory consistency with the thread that
        // will execute this task is guaranteed.
        // Note that the state of this object will not be touched after this line (-> tail recursion).
        executorService.schedule(this, delay, TimeUnit.MILLISECONDS);
        returnValue = true;
    }
    return returnValue;
}

private static ClientConfiguration withRetryPolicy(ClientConfiguration config, String tableName, String indexName) {
    RetryPolicy.RetryCondition retryCondition = new PredefinedRetryPolicies.SDKDefaultRetryCondition() {
        @Override
        public boolean shouldRetry(AmazonWebServiceRequest originalRequest,
                                   AmazonClientException exception,
                                   int retriesAttempted)
        {
            if (exception instanceof AmazonServiceException) {
                AmazonServiceException ase = (AmazonServiceException)exception;
                if (RetryUtils.isThrottlingException(ase)) {
                    if ( null != indexName ) {
                        if(LOG.isWarnEnabled())
                            LOG.warn("throttling on table "+tableName+" index "+indexName);
                    } else {
                        if(LOG.isWarnEnabled())
                            LOG.warn("throttling on table "+tableName+" (main index)");
                    }
                }
            }
            return super.shouldRetry(originalRequest, exception, retriesAttempted);
        }
    };
    config.withRetryPolicy(new RetryPolicy(
                               retryCondition,
                               PredefinedRetryPolicies.DYNAMODB_DEFAULT_BACKOFF_STRATEGY,
                               PredefinedRetryPolicies.DYNAMODB_DEFAULT_MAX_ERROR_RETRY,
                               true));
    return config;
}

protected AmazonApiGateway getApiGatewayClient() {
    if (apiGatewayClient != null) {
        return apiGatewayClient;
    }

    RetryPolicy.RetryCondition retryCondition = new RetryPolicy.RetryCondition() {

        @Override
        public boolean shouldRetry(AmazonWebServiceRequest amazonWebServiceRequest, AmazonClientException amazonClientException, int i) {
            if (amazonClientException instanceof TooManyRequestsException) {
                return true;
            }
            return PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION.shouldRetry(amazonWebServiceRequest,
                    amazonClientException, i);
        }
    };

    RetryPolicy retryPolicy = new RetryPolicy(retryCondition,
            PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY,
            10, true);

    ClientConfiguration clientConfig = new ClientConfiguration()
            .withRetryPolicy(retryPolicy);

    apiGatewayClient = new AmazonApiGatewayClient(getAWSCredentialsProvideChain(), clientConfig).withRegion(Region.getRegion(Regions.fromName(deployment.getRegion())));
    return apiGatewayClient;
}

private AmazonKinesisAsyncClient createClient() {

        // Building Kinesis configuration
        int connectionTimeout = ClientConfiguration.DEFAULT_CONNECTION_TIMEOUT;
        int maxConnection = ClientConfiguration.DEFAULT_MAX_CONNECTIONS;

        RetryPolicy retryPolicy = ClientConfiguration.DEFAULT_RETRY_POLICY;
        int socketTimeout = ClientConfiguration.DEFAULT_SOCKET_TIMEOUT;
        boolean useReaper = ClientConfiguration.DEFAULT_USE_REAPER;
        String userAgent = ClientConfiguration.DEFAULT_USER_AGENT;

        ClientConfiguration clientConfiguration = new ClientConfiguration();
        clientConfiguration.setConnectionTimeout(connectionTimeout);
        clientConfiguration.setMaxConnections(maxConnection);
        clientConfiguration.setRetryPolicy(retryPolicy);
        clientConfiguration.setSocketTimeout(socketTimeout);
        clientConfiguration.setUseReaper(useReaper);
        clientConfiguration.setUserAgent(userAgent);

        // Reading credentials from ENV-variables
        AWSCredentialsProvider awsCredentialsProvider = new DefaultAWSCredentialsProviderChain();

        // Configuring Kinesis-client with configuration
        AmazonKinesisAsyncClient kinesisAsyncClient = new AmazonKinesisAsyncClient(awsCredentialsProvider, clientConfiguration);
        Regions myRegion = Regions.fromName(AmazonUtil.getInstance().getRegion());
        kinesisAsyncClient.withRegion(Region.getRegion(myRegion));

        return kinesisAsyncClient;
    }

@Provides
protected ApiGateway provideAmazonApiGateway(AWSCredentialsProvider credsProvider,
                                             RetryPolicy.BackoffStrategy backoffStrategy,
                                             @Named("region") String region) {

    final RetryPolicy retrypolicy = new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, backoffStrategy, 5, true);

    final ClientConfiguration clientConfig = new ClientConfiguration().withUserAgent(USER_AGENT).withRetryPolicy(retrypolicy);

    return new AmazonApiGateway(getEndpoint(region)).with(credsProvider).with(clientConfig).getApiGateway();
}

@Provides
protected RetryPolicy.BackoffStrategy provideBackoffStrategy() {

    // tune these parameters to handle throttling errors
    final int maxBackoffInMilliseconds = 50 * 1000; // maximum exponential back-off time before retrying a request
    final int throttlingScaleFactor = 800; // base sleep time for throttling exceptions
    final int maxRetriesBeforeBackoff = 10; // log2(maxBackoffInMilliseconds/throttlingScaleFactor)

    final int baseScaleFactor = 600; // base sleep time for general exceptions
    final int throttlingScaleFactorRandomRange = throttlingScaleFactor / 4;

    final Random random = new Random();

    return (originalRequest, exception, retriesAttempted) -> {

        LOG.debug("Caught error from service. Retry attempt: " + retriesAttempted, exception);

        if (retriesAttempted < 0) return 0;
        if (retriesAttempted > maxRetriesBeforeBackoff) return maxBackoffInMilliseconds;

        int scaleFactor;
        if (exception instanceof AmazonServiceException
                && RetryUtils.isThrottlingException((AmazonServiceException) exception)) {
            scaleFactor = throttlingScaleFactor + random.nextInt(throttlingScaleFactorRandomRange);
        } else {
            scaleFactor = baseScaleFactor;
        }

        long delay = (1L << retriesAttempted) * scaleFactor;
        delay = Math.min(delay, maxBackoffInMilliseconds);

        LOG.info("Client backing off for " + delay + "ms");

        return delay;
    };
}

/**
 * Asserts that the retry policy generated by the factory is configured with the correct values.
 */
@Test
public void assertResultRetryPolicyConfiguredCorrectly()
{
    int expectedMaxErrorRetry = 12345;

    when(configurationHelper.getProperty(any(), eq(Integer.class))).thenReturn(expectedMaxErrorRetry);

    RetryPolicy retryPolicy = retryPolicyFactory.getRetryPolicy();

    assertEquals(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, retryPolicy.getRetryCondition());
    assertEquals(backoffStrategy, retryPolicy.getBackoffStrategy());
    assertEquals(expectedMaxErrorRetry, retryPolicy.getMaxErrorRetry());
}

@Test
public void testDeleteDirectoryNoS3VersionsExist()
{
    // Create an S3 file transfer request parameters DTO to access S3 objects.
    S3FileTransferRequestParamsDto s3FileTransferRequestParamsDto = new S3FileTransferRequestParamsDto();
    s3FileTransferRequestParamsDto.setS3BucketName(S3_BUCKET_NAME);
    s3FileTransferRequestParamsDto.setS3KeyPrefix(S3_KEY_PREFIX);

    // Create a retry policy.
    RetryPolicy retryPolicy =
        new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY, INTEGER_VALUE, true);

    // Create an empty version listing.
    VersionListing versionListing = new VersionListing();

    // Mock the external calls.
    when(retryPolicyFactory.getRetryPolicy()).thenReturn(retryPolicy);
    when(s3Operations.listVersions(any(ListVersionsRequest.class), any(AmazonS3Client.class))).thenReturn(versionListing);

    // Call the method under test.
    s3DaoImpl.deleteDirectory(s3FileTransferRequestParamsDto);

    // Verify the external calls.
    verify(retryPolicyFactory).getRetryPolicy();
    verify(s3Operations).listVersions(any(ListVersionsRequest.class), any(AmazonS3Client.class));
    verifyNoMoreInteractionsHelper();
}

@Provides
protected ApiGateway provideAmazonApiGateway(AWSCredentialsProvider credsProvider,
                                             RetryPolicy.BackoffStrategy backoffStrategy,
                                             @Named("region") String region) {

    final RetryPolicy retrypolicy = new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, backoffStrategy, 5, true);

    final ClientConfiguration clientConfig = new ClientConfiguration().withUserAgent(USER_AGENT).withRetryPolicy(retrypolicy);

    return new AmazonApiGateway(getEndpoint(region)).with(credsProvider).with(clientConfig).getApiGateway();
}

@Provides
protected RetryPolicy.BackoffStrategy provideBackoffStrategy() {

    // tune these parameters to handle throttling errors
    final int maxBackoffInMilliseconds = 50 * 1000; // maximum exponential back-off time before retrying a request
    final int throttlingScaleFactor = 800; // base sleep time for throttling exceptions
    final int maxRetriesBeforeBackoff = 10; // log2(maxBackoffInMilliseconds/throttlingScaleFactor)

    final int baseScaleFactor = 600; // base sleep time for general exceptions
    final int throttlingScaleFactorRandomRange = throttlingScaleFactor / 4;

    final Random random = new Random();

    return (originalRequest, exception, retriesAttempted) -> {

        LOG.debug("Caught error from service. Retry attempt: " + retriesAttempted, exception);

        if (retriesAttempted < 0) return 0;
        if (retriesAttempted > maxRetriesBeforeBackoff) return maxBackoffInMilliseconds;

        int scaleFactor;
        if (exception instanceof AmazonServiceException
                && RetryUtils.isThrottlingException((AmazonServiceException) exception)) {
            scaleFactor = throttlingScaleFactor + random.nextInt(throttlingScaleFactorRandomRange);
        } else {
            scaleFactor = baseScaleFactor;
        }

        long delay = (1L << retriesAttempted) * scaleFactor;
        delay = Math.min(delay, maxBackoffInMilliseconds);

        LOG.info("Client backing off for " + delay + "ms");

        return delay;
    };
}

protected static ClientConfiguration buildConfiguration(Logger logger, Settings settings) {
    ClientConfiguration clientConfiguration = new ClientConfiguration();
    // the response metadata cache is only there for diagnostics purposes,
    // but can force objects from every response to the old generation.
    clientConfiguration.setResponseMetadataCacheSize(0);
    clientConfiguration.setProtocol(CLOUD_EC2.PROTOCOL_SETTING.get(settings));

    if (PROXY_HOST_SETTING.exists(settings) || CLOUD_EC2.PROXY_HOST_SETTING.exists(settings)) {
        String proxyHost = CLOUD_EC2.PROXY_HOST_SETTING.get(settings);
        Integer proxyPort = CLOUD_EC2.PROXY_PORT_SETTING.get(settings);
        String proxyUsername = CLOUD_EC2.PROXY_USERNAME_SETTING.get(settings);
        String proxyPassword = CLOUD_EC2.PROXY_PASSWORD_SETTING.get(settings);

        clientConfiguration
            .withProxyHost(proxyHost)
            .withProxyPort(proxyPort)
            .withProxyUsername(proxyUsername)
            .withProxyPassword(proxyPassword);
    }

    // #155: we might have 3rd party users using older EC2 API version
    String awsSigner = CLOUD_EC2.SIGNER_SETTING.get(settings);
    if (Strings.hasText(awsSigner)) {
        logger.debug("using AWS API signer [{}]", awsSigner);
        AwsSigner.configureSigner(awsSigner, clientConfiguration);
    }

    // Increase the number of retries in case of 5xx API responses
    final Random rand = Randomness.get();
    RetryPolicy retryPolicy = new RetryPolicy(
        RetryPolicy.RetryCondition.NO_RETRY_CONDITION,
        new RetryPolicy.BackoffStrategy() {
            @Override
            public long delayBeforeNextRetry(AmazonWebServiceRequest originalRequest,
                                             AmazonClientException exception,
                                             int retriesAttempted) {
                // with 10 retries the max delay time is 320s/320000ms (10 * 2^5 * 1 * 1000)
                logger.warn("EC2 API request failed, retry again. Reason was:", exception);
                return 1000L * (long) (10d * Math.pow(2, retriesAttempted / 2.0d) * (1.0d + rand.nextDouble()));
            }
        },
        10,
        false);
    clientConfiguration.setRetryPolicy(retryPolicy);
    clientConfiguration.setSocketTimeout((int) CLOUD_EC2.READ_TIMEOUT.get(settings).millis());

    return clientConfiguration;
}

public RetryPolicy buildRetryPolicy() {
    RetryPolicy policy = new RetryPolicy(new AlwaysRetryCondition(), new SimpleArrayBackoffStrategy(BACKOFF_VALUES),
            3, false);
    return policy;
}

@Test
@Ignore
public void clientConfigurationCopyConstructor_CopiesAllValues() throws Exception {
    ClientConfiguration customConfig = new ClientConfiguration();

    for (Field field : ClientConfiguration.class.getDeclaredFields()) {
        if (isStaticField(field)) {
            continue;
        }
        field.setAccessible(true);
        final Class<?> clzz = field.getType();

        if (clzz.isAssignableFrom(int.class) || clzz.isAssignableFrom(long.class)) {
            field.set(customConfig, Math.abs(RANDOM.nextInt()));
        } else if (clzz.isAssignableFrom(boolean.class)) {
            // Invert the default value to ensure it's different
            field.set(customConfig, !(Boolean) field.get(customConfig));
        } else if (clzz.isAssignableFrom(String.class)) {
            field.set(customConfig, RandomStringUtils.random(10));
        } else if (clzz.isAssignableFrom(RetryPolicy.class)) {
            field.set(customConfig, CUSTOM_RETRY_POLICY);
        } else if (clzz.isAssignableFrom(InetAddress.class)) {
            field.set(customConfig, InetAddress.getLocalHost());
        } else if (clzz.isAssignableFrom(Protocol.class)) {
            // Default is HTTPS so switch to HTTP
            field.set(customConfig, Protocol.HTTP);
        } else if (clzz.isAssignableFrom(DnsResolver.class)) {
            field.set(customConfig, new MyCustomDnsResolver());
        } else if (clzz.isAssignableFrom(SecureRandom.class)) {
            field.set(customConfig, new SecureRandom());
        } else if (field.getName().equals("headers")) {
            field.set(customConfig, ImmutableMapParameter.of("foo", "bar"));
        } else if (clzz.isAssignableFrom(ApacheHttpClientConfig.class)) {
            customConfig.getApacheHttpClientConfig()
                        .setSslSocketFactory(Mockito.mock(ConnectionSocketFactory.class));
        } else if (clzz.isAssignableFrom(List.class)) {
            field.set(customConfig, new ArrayList<Object>());
        } else {
            throw new RuntimeException(
                    String.format("Field %s of type %s is not supported",
                                  field.getName(),
                                  field.getType()));
        }
        // Extra check to make sure the value differs from the default and we haven't missed something
        assertNotEquals(
                String.format("Field %s does not differ from default value", field.getName()),
                field.get(DEFAULT_CLIENT_CONFIG), field.get(customConfig));
    }

    // Do a deep comparison of the config after sending it through the copy constructor
    assertReflectionEquals(customConfig, new ClientConfiguration(customConfig));
}

/**
 * Configures appender instance and makes it ready for use by the consumers.
 * It validates mandatory parameters and confirms if the configured stream is
 * ready for publishing data yet.
 * 
 * Error details are made available through the fallback handler for this
 * appender
 * 
 * @throws IllegalStateException if we encounter issues configuring this
 *           appender instance
 */
@Override
public void start() {
  if(layout == null) {
    initializationFailed = true;
    addError("Invalid configuration - No layout for appender: " + name);
    return;
  }

  if(streamName == null) {
    initializationFailed = true;
    addError("Invalid configuration - streamName cannot be null for appender: " + name);
    return;
  }

  ClientConfiguration clientConfiguration = new ClientConfiguration();
  clientConfiguration.setMaxErrorRetry(maxRetries);
  clientConfiguration
      .setRetryPolicy(new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION,
                                      PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY, maxRetries, true));
  clientConfiguration.setUserAgent(AppenderConstants.USER_AGENT_STRING);

  BlockingQueue<Runnable> taskBuffer = new LinkedBlockingDeque<Runnable>(bufferSize);
  threadPoolExecutor = new ThreadPoolExecutor(threadCount, threadCount,
                                              AppenderConstants.DEFAULT_THREAD_KEEP_ALIVE_SEC, TimeUnit.SECONDS,
                                              taskBuffer, new BlockFastProducerPolicy());
  threadPoolExecutor.prestartAllCoreThreads();

  this.client = createClient(credentials, clientConfiguration, threadPoolExecutor);

  client.setRegion(findRegion());
  if(!Validator.isBlank(endpoint)) {
    if(!Validator.isBlank(region)) {
      addError("Received configuration for both region as well as Amazon Kinesis endpoint. (" + endpoint
               + ") will be used as endpoint instead of default endpoint for region (" + region + ")");
    }
    client.setEndpoint(endpoint);
  }

  validateStreamName(client, streamName);

  super.start();
}

@Test
public void testGetTemporarySecurityCredentials()
{
    // Create an AWS parameters DTO with proxy settings.
    AwsParamsDto awsParamsDto = new AwsParamsDto();
    awsParamsDto.setHttpProxyHost(HTTP_PROXY_HOST);
    awsParamsDto.setHttpProxyPort(HTTP_PROXY_PORT);

    // Specify the duration, in seconds, of the role session.
    int awsRoleDurationSeconds = INTEGER_VALUE;

    // Create an IAM policy.
    Policy policy = new Policy(STRING_VALUE);

    // Create a retry policy.
    RetryPolicy retryPolicy =
        new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY, INTEGER_VALUE, true);

    // Create the expected assume role request.
    AssumeRoleRequest assumeRoleRequest = new AssumeRoleRequest().withRoleArn(AWS_ROLE_ARN).withRoleSessionName(SESSION_NAME).withPolicy(policy.toJson())
        .withDurationSeconds(awsRoleDurationSeconds);

    // Create AWS credentials for API authentication.
    Credentials credentials = new Credentials();
    credentials.setAccessKeyId(AWS_ASSUMED_ROLE_ACCESS_KEY);
    credentials.setSecretAccessKey(AWS_ASSUMED_ROLE_SECRET_KEY);
    credentials.setSessionToken(AWS_ASSUMED_ROLE_SESSION_TOKEN);

    // Create an assume role result.
    AssumeRoleResult assumeRoleResult = new AssumeRoleResult();
    assumeRoleResult.setCredentials(credentials);

    // Mock the external calls.
    when(retryPolicyFactory.getRetryPolicy()).thenReturn(retryPolicy);
    when(stsOperations.assumeRole(any(AWSSecurityTokenServiceClient.class), eq(assumeRoleRequest))).thenReturn(assumeRoleResult);

    // Call the method under test.
    Credentials result = stsDaoImpl.getTemporarySecurityCredentials(awsParamsDto, SESSION_NAME, AWS_ROLE_ARN, awsRoleDurationSeconds, policy);

    // Verify the external calls.
    verify(retryPolicyFactory).getRetryPolicy();
    verify(stsOperations).assumeRole(any(AWSSecurityTokenServiceClient.class), eq(assumeRoleRequest));
    verifyNoMoreInteractionsHelper();

    // Validate the returned object.
    assertEquals(credentials, result);
}

@Test
public void testGetTemporarySecurityCredentialsMissingOptionalParameters()
{
    // Create an AWS parameters DTO without proxy settings.
    AwsParamsDto awsParamsDto = new AwsParamsDto();

    // Specify the duration, in seconds, of the role session.
    int awsRoleDurationSeconds = INTEGER_VALUE;

    // Create a retry policy.
    RetryPolicy retryPolicy =
        new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY, INTEGER_VALUE, true);

    // Create the expected assume role request.
    AssumeRoleRequest assumeRoleRequest =
        new AssumeRoleRequest().withRoleArn(AWS_ROLE_ARN).withRoleSessionName(SESSION_NAME).withDurationSeconds(awsRoleDurationSeconds);

    // Create AWS credentials for API authentication.
    Credentials credentials = new Credentials();
    credentials.setAccessKeyId(AWS_ASSUMED_ROLE_ACCESS_KEY);
    credentials.setSecretAccessKey(AWS_ASSUMED_ROLE_SECRET_KEY);
    credentials.setSessionToken(AWS_ASSUMED_ROLE_SESSION_TOKEN);

    // Create an assume role result.
    AssumeRoleResult assumeRoleResult = new AssumeRoleResult();
    assumeRoleResult.setCredentials(credentials);

    // Mock the external calls.
    when(retryPolicyFactory.getRetryPolicy()).thenReturn(retryPolicy);
    when(stsOperations.assumeRole(any(AWSSecurityTokenServiceClient.class), eq(assumeRoleRequest))).thenReturn(assumeRoleResult);

    // Call the method under test. Please note that we do not specify an IAM policy.
    Credentials result = stsDaoImpl.getTemporarySecurityCredentials(awsParamsDto, SESSION_NAME, AWS_ROLE_ARN, awsRoleDurationSeconds, null);

    // Verify the external calls.
    verify(retryPolicyFactory).getRetryPolicy();
    verify(stsOperations).assumeRole(any(AWSSecurityTokenServiceClient.class), eq(assumeRoleRequest));
    verifyNoMoreInteractionsHelper();

    // Validate the returned object.
    assertEquals(credentials, result);
}

@Test
public void testDeleteDirectoryMultiObjectDeleteException()
{
    // Create an S3 file transfer request parameters DTO to access S3 objects.
    S3FileTransferRequestParamsDto s3FileTransferRequestParamsDto = new S3FileTransferRequestParamsDto();
    s3FileTransferRequestParamsDto.setS3BucketName(S3_BUCKET_NAME);
    s3FileTransferRequestParamsDto.setS3KeyPrefix(S3_KEY_PREFIX);

    // Create a retry policy.
    RetryPolicy retryPolicy =
        new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY, INTEGER_VALUE, true);

    // Create an S3 version summary.
    S3VersionSummary s3VersionSummary = new S3VersionSummary();
    s3VersionSummary.setKey(S3_KEY);
    s3VersionSummary.setVersionId(S3_VERSION_ID);

    // Create a version listing.
    VersionListing versionListing = new VersionListing();
    versionListing.setVersionSummaries(Arrays.asList(s3VersionSummary));

    // Create a delete error.
    MultiObjectDeleteException.DeleteError deleteError = new MultiObjectDeleteException.DeleteError();
    deleteError.setKey(S3_KEY);
    deleteError.setVersionId(S3_VERSION_ID);
    deleteError.setCode(ERROR_CODE);
    deleteError.setMessage(ERROR_MESSAGE);

    // Create a multi object delete exception.
    MultiObjectDeleteException multiObjectDeleteException = new MultiObjectDeleteException(Arrays.asList(deleteError), new ArrayList<>());

    // Mock the external calls.
    when(retryPolicyFactory.getRetryPolicy()).thenReturn(retryPolicy);
    when(s3Operations.listVersions(any(ListVersionsRequest.class), any(AmazonS3Client.class))).thenReturn(versionListing);
    when(s3Operations.deleteObjects(any(DeleteObjectsRequest.class), any(AmazonS3Client.class))).thenThrow(multiObjectDeleteException);

    // Try to call the method under test.
    try
    {
        s3DaoImpl.deleteDirectory(s3FileTransferRequestParamsDto);
    }
    catch (IllegalStateException e)
    {
        assertEquals(String.format("Failed to delete keys/key versions with prefix \"%s\" from bucket \"%s\". " +
                "Reason: One or more objects could not be deleted (Service: null; Status Code: 0; Error Code: null; Request ID: null; S3 Extended Request ID: null)",
            S3_KEY_PREFIX, S3_BUCKET_NAME), e.getMessage());
    }

    // Verify the external calls.
    verify(retryPolicyFactory, times(2)).getRetryPolicy();
    verify(s3Operations).listVersions(any(ListVersionsRequest.class), any(AmazonS3Client.class));
    verify(s3Operations).deleteObjects(any(DeleteObjectsRequest.class), any(AmazonS3Client.class));
    verifyNoMoreInteractionsHelper();
}

@Test
public void testTagObjects()
{
    // Create an S3 file transfer request parameters DTO to access S3 objects.
    S3FileTransferRequestParamsDto s3FileTransferRequestParamsDto = new S3FileTransferRequestParamsDto();
    s3FileTransferRequestParamsDto.setS3BucketName(S3_BUCKET_NAME);
    s3FileTransferRequestParamsDto.setFiles(Arrays.asList(new File(S3_KEY_PREFIX + "/" + LOCAL_FILE)));

    // Create an S3 file transfer request parameters DTO to tag S3 objects.
    S3FileTransferRequestParamsDto s3ObjectTaggerParamsDto = new S3FileTransferRequestParamsDto();
    s3ObjectTaggerParamsDto.setAwsAccessKeyId(AWS_ASSUMED_ROLE_ACCESS_KEY);
    s3ObjectTaggerParamsDto.setAwsSecretKey(AWS_ASSUMED_ROLE_SECRET_KEY);
    s3ObjectTaggerParamsDto.setSessionToken(AWS_ASSUMED_ROLE_SESSION_TOKEN);

    // Create an S3 object tag.
    Tag tag = new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE);

    // Create a retry policy.
    RetryPolicy retryPolicy =
        new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY, INTEGER_VALUE, true);

    // Create a get object tagging result.
    GetObjectTaggingResult getObjectTaggingResult = new GetObjectTaggingResult(null);

    // Create a set object tagging result.
    SetObjectTaggingResult setObjectTaggingResult = new SetObjectTaggingResult();

    // Mock the external calls.
    when(retryPolicyFactory.getRetryPolicy()).thenReturn(retryPolicy);
    when(s3Operations.getObjectTagging(any(GetObjectTaggingRequest.class), any(AmazonS3Client.class))).thenReturn(getObjectTaggingResult);
    when(s3Operations.setObjectTagging(any(SetObjectTaggingRequest.class), any(AmazonS3Client.class))).thenReturn(setObjectTaggingResult);

    // Call the method under test.
    s3DaoImpl.tagObjects(s3FileTransferRequestParamsDto, s3ObjectTaggerParamsDto, tag);

    // Verify the external calls.
    verify(retryPolicyFactory, times(2)).getRetryPolicy();
    verify(s3Operations).getObjectTagging(any(GetObjectTaggingRequest.class), any(AmazonS3Client.class));
    verify(s3Operations).setObjectTagging(any(SetObjectTaggingRequest.class), any(AmazonS3Client.class));
    verifyNoMoreInteractionsHelper();
}

private AmazonKinesisAsyncClient createClient() {

        // Building Kinesis configuration
        int connectionTimeout = getOptionalIntConfig(CONNECTION_TIMEOUT, ClientConfiguration.DEFAULT_CONNECTION_TIMEOUT);
        int maxConnection = getOptionalIntConfig(MAX_CONNECTION, ClientConfiguration.DEFAULT_MAX_CONNECTIONS);

        // TODO: replace default retry policy
        RetryPolicy retryPolicy = ClientConfiguration.DEFAULT_RETRY_POLICY;
        int socketTimeout = getOptionalIntConfig(SOCKET_TIMEOUT, ClientConfiguration.DEFAULT_SOCKET_TIMEOUT);
        boolean useReaper = getOptionalBooleanConfig(USE_REAPER, ClientConfiguration.DEFAULT_USE_REAPER);
        String userAgent = getOptionalStringConfig(USER_AGENT, ClientConfiguration.DEFAULT_USER_AGENT);
        String endpoint = getOptionalStringConfig(ENDPOINT, null);


        streamName = getMandatoryStringConfig(STREAM_NAME);
        partitionKey = getMandatoryStringConfig(PARTITION_KEY);
        region = getMandatoryStringConfig(REGION);

        logger.info(" --- Stream name: " + streamName);
        logger.info(" --- Partition key: " + partitionKey);
        logger.info(" --- Region: " + region);
        if(endpoint != null) {
            logger.info(" --- Endpoint: " + endpoint);
        }

        ClientConfiguration clientConfiguration = new ClientConfiguration();
        clientConfiguration.setConnectionTimeout(connectionTimeout);
        clientConfiguration.setMaxConnections(maxConnection);
        clientConfiguration.setRetryPolicy(retryPolicy);
        clientConfiguration.setSocketTimeout(socketTimeout);
        clientConfiguration.setUseReaper(useReaper);
        clientConfiguration.setUserAgent(userAgent);

        /*
        AWS credentials provider chain that looks for credentials in this order:
            Environment Variables - AWS_ACCESS_KEY_ID and AWS_SECRET_KEY
            Java System Properties - aws.accessKeyId and aws.secretKey
            Credential profiles file at the default location (~/.aws/credentials) shared by all AWS SDKs and the AWS CLI
            Instance profile credentials delivered through the Amazon EC2 metadata service
        */

        AWSCredentialsProvider awsCredentialsProvider = new DefaultAWSCredentialsProviderChain();

        // Configuring Kinesis-client with configuration
        AmazonKinesisAsyncClient kinesisAsyncClient = new AmazonKinesisAsyncClient(awsCredentialsProvider, clientConfiguration);
        Region awsRegion = RegionUtils.getRegion(region);
        kinesisAsyncClient.setRegion(awsRegion);
        if(endpoint != null) {
            kinesisAsyncClient.setEndpoint(endpoint);
        }

        return kinesisAsyncClient;
    }

@Bean
RetryPolicy.RetryCondition instrumentedRetryCondition(Registry registry) {
  return new InstrumentedRetryCondition(registry);
}

@Bean
RetryPolicy.BackoffStrategy instrumentedBackoffStrategy(Registry registry) {
  return new InstrumentedBackoffStrategy(registry);
}

public InstrumentedRetryCondition(Registry registry, RetryPolicy.RetryCondition delegate) {
  this.registry = Objects.requireNonNull(registry, "registry");
  this.delegate = Objects.requireNonNull(delegate, "delegate");
}

public InstrumentedBackoffStrategy(Registry registry, RetryPolicy.BackoffStrategy delegate) {
  this.registry = Objects.requireNonNull(registry, "registry");
  this.delegate = Objects.requireNonNull(delegate, "delegate");
}

/**
  * Configures this appender instance and makes it ready for use by the
  * consumers. It validates mandatory parameters and confirms if the configured
  * stream is ready for publishing data yet.
  * 
  * Error details are made available through the fallback handler for this
  * appender
  * 
  * @throws IllegalStateException
  *           if we encounter issues configuring this appender instance
  */
 @Override
 public void activateOptions() {
   if (streamName == null) {
     initializationFailed = true;
     error("Invalid configuration - streamName cannot be null for appender: " + name);
   }

   if (layout == null) {
     initializationFailed = true;
     error("Invalid configuration - No layout for appender: " + name);
   }

   ClientConfiguration clientConfiguration = new ClientConfiguration();
   clientConfiguration = setProxySettingsFromSystemProperties(clientConfiguration);

   clientConfiguration.setMaxErrorRetry(maxRetries);
   clientConfiguration.setRetryPolicy(new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION,
       PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY, maxRetries, true));
   clientConfiguration.setUserAgent(AppenderConstants.USER_AGENT_STRING);

   BlockingQueue<Runnable> taskBuffer = new LinkedBlockingDeque<Runnable>(bufferSize);
   ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(threadCount, threadCount,
       AppenderConstants.DEFAULT_THREAD_KEEP_ALIVE_SEC, TimeUnit.SECONDS, taskBuffer, new BlockFastProducerPolicy());
   threadPoolExecutor.prestartAllCoreThreads();
   kinesisClient = new AmazonKinesisAsyncClient(new CustomCredentialsProviderChain(), clientConfiguration,
       threadPoolExecutor);

   boolean regionProvided = !Validator.isBlank(region);
   if (!regionProvided) {
     region = AppenderConstants.DEFAULT_REGION;
   }
   if (!Validator.isBlank(endpoint)) {
     if (regionProvided) {
LOGGER
    .warn("Received configuration for both region as well as Amazon Kinesis endpoint. ("
    + endpoint
    + ") will be used as endpoint instead of default endpoint for region ("
    + region + ")");
     }
     kinesisClient.setEndpoint(endpoint,
  AppenderConstants.DEFAULT_SERVICE_NAME, region);
   } else {
     kinesisClient.setRegion(Region.getRegion(Regions.fromName(region)));
   }

   DescribeStreamResult describeResult = null;
   try {
     describeResult = kinesisClient.describeStream(streamName);
     String streamStatus = describeResult.getStreamDescription().getStreamStatus();
     if (!StreamStatus.ACTIVE.name().equals(streamStatus) && !StreamStatus.UPDATING.name().equals(streamStatus)) {
       initializationFailed = true;
       error("Stream " + streamName + " is not ready (in active/updating status) for appender: " + name);
     }
   } catch (ResourceNotFoundException rnfe) {
     initializationFailed = true;
     error("Stream " + streamName + " doesn't exist for appender: " + name, rnfe);
   }

   asyncCallHander = new AsyncPutCallStatsReporter(name);
 }

public AWSDatabaseHolder(Config config) {
    maxAgeInMs = config.getDuration("maxAge", TimeUnit.MILLISECONDS);

    final DefaultAWSCredentialsProviderChain awsCredentialsProviderChain = new DefaultAWSCredentialsProviderChain();

    final ClientConfiguration clientConfig = new ClientConfiguration();
    clientConfig.setRetryPolicy(new RetryPolicy(null, null, config.getInt("maxErrorRetry"), true));

    final AmazonEC2Client bootstrapEC2Client = new AmazonEC2Client(awsCredentialsProviderChain);
    ec2Clients = Maps.newHashMap();
    rdsClients = Maps.newHashMap();
    sqsClients = Maps.newHashMap();
    dynamoDBClients = Maps.newHashMap();
    elasticacheClients = Maps.newHashMap();

    final List<Region> ec2Regions = bootstrapEC2Client.describeRegions().getRegions();
    for (Region region : ec2Regions) {
        final String regionName = region.getRegionName();
        final String endpoint = region.getEndpoint();
        log.debug("Adding ec2 region {}", region);

        final AmazonEC2Client ec2Client = new AmazonEC2Client(awsCredentialsProviderChain, clientConfig);
        ec2Client.setEndpoint(endpoint);
        ec2Clients.put(regionName, ec2Client);

        final AmazonRDSClient rdsClient = new AmazonRDSClient(awsCredentialsProviderChain, clientConfig);
        rdsClient.setEndpoint(endpoint.replaceFirst("ec2\\.", "rds."));
        rdsClients.put(regionName, rdsClient);

        final AmazonDynamoDBClient dynamoDBClient =
            new AmazonDynamoDBClient(awsCredentialsProviderChain, clientConfig);
        dynamoDBClient.setEndpoint(endpoint.replaceFirst("ec2\\.", "dynamodb."));
        dynamoDBClients.put(regionName, dynamoDBClient);

        final AmazonSQSClient sqsClient = new AmazonSQSClient(awsCredentialsProviderChain, clientConfig);
        sqsClient.setEndpoint(endpoint.replaceFirst("ec2\\.", "sqs."));
        sqsClients.put(regionName, sqsClient);

        final AmazonElastiCacheClient elastiCacheClient = new AmazonElastiCacheClient
            (awsCredentialsProviderChain, clientConfig);
        elastiCacheClient.setEndpoint(endpoint.replaceFirst("ec2\\.", "elasticache."));
        elasticacheClients.put(regionName, elastiCacheClient);
    }

    this.iamClient = new AmazonIdentityManagementClient(awsCredentialsProviderChain, clientConfig);

    if (config.hasPath("accountNumber")) {
        this.awsAccountNumber = config.getString("accountNumber");
    } else {
        this.awsAccountNumber = null;
    }

    rebuild();
}

/**
 * Returns the retry policy upon failed requests.
 *
 * @return The retry policy upon failed requests.
 */
public RetryPolicy getRetryPolicy() {
    return retryPolicy;
}

/**
 * Sets the retry policy upon failed requests. User could specify whether the RetryPolicy should
 * honor maxErrorRetry set by {@link #setMaxErrorRetry(int)}.
 *
 * @param retryPolicy
 *            The retry policy upon failed requests.
 */
public void setRetryPolicy(RetryPolicy retryPolicy) {
    this.retryPolicy = retryPolicy;
}

/**
 * Sets the retry policy upon failed requests, and returns the updated ClientConfiguration
 * object. User could specify whether the RetryPolicy should honor maxErrorRetry set by
 * {@link #setMaxErrorRetry(int)}
 *
 * @param retryPolicy
 *            The retry policy upon failed requests.
 */
public ClientConfiguration withRetryPolicy(RetryPolicy retryPolicy) {
    setRetryPolicy(retryPolicy);
    return this;
}

/**
 * Gets the application's default retry policy. The policy uses PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, a SimpleExponentialBackoffStrategy, and the
 * maximum number of attempts is dynamically configurable through AWS_MAX_RETRY_ATTEMPT.
 * 
 * @return RetryPolicy
 */
public RetryPolicy getRetryPolicy()
{
    return new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, backoffStrategy,
        configurationHelper.getProperty(ConfigurationValue.AWS_MAX_RETRY_ATTEMPT, Integer.class), true);
}