Java 类com.amazonaws.handlers.AsyncHandler 实例源码

private void sendMetricData(List<MetricDatum> metricData) {
    PutMetricDataRequest putMetricDataRequest = new PutMetricDataRequest()
            .withNamespace(config.namespace())
            .withMetricData(metricData);
    amazonCloudWatchAsync.putMetricDataAsync(putMetricDataRequest, new AsyncHandler<PutMetricDataRequest, PutMetricDataResult>() {
        @Override
        public void onError(Exception exception) {
            logger.error("Error sending metric data.", exception);
        }

        @Override
        public void onSuccess(PutMetricDataRequest request, PutMetricDataResult result) {
            logger.debug("Published metric with namespace:{}", request.getNamespace());
        }
    });
}

@Override
public void process(AmazonEC2Async amazonEC2Async, Instance instance) {
    amazonEC2Async.stopInstancesAsync(
            new StopInstancesRequest().withInstanceIds(instance.getInstanceId()),
            new AsyncHandler<StopInstancesRequest, StopInstancesResult>() {
                @Override
                public void onError(Exception exception) {
                    log.warn("something went wrong stopping the server {}",
                            exception.getLocalizedMessage());
                }

                @Override
                public void onSuccess(StopInstancesRequest request, StopInstancesResult result) {
                    onSuccessStop(instance);
                }
            });
}

@Override
public void process(Instance instance) {
    ec2UserClient.amazonEC2Async().stopInstancesAsync(
            new StopInstancesRequest().withInstanceIds(instance.getInstanceId()),
            new AsyncHandler<StopInstancesRequest, StopInstancesResult>() {
                @Override
                public void onError(Exception exception) {
                    log.warn("something went wrong stopping the server {}",
                            exception.getLocalizedMessage());
                }

                @Override
                public void onSuccess(StopInstancesRequest request, StopInstancesResult result) {
                    onSuccessStop(instance);
                }
            });
}

@Override
public void process(Instance instance) {
    if (instance.getKeyName().contains("swordfish-")) {
        ec2KeyPair.delete(instance);
    }

    ec2UserClient.amazonEC2Async().terminateInstancesAsync(
            new TerminateInstancesRequest().withInstanceIds(instance.getInstanceId()),
            new AsyncHandler<TerminateInstancesRequest, TerminateInstancesResult>() {
                @Override
                public void onError(Exception exception) {
                    log.warn("something went wrong terminating the server {}",
                            exception.getLocalizedMessage());
                }

                @Override
                public void onSuccess(TerminateInstancesRequest request, TerminateInstancesResult result) {
                    ec2UserClient.amazonEC2Async().waiters()
                            .instanceRunning()
                            .runAsync(ec2Waiter.describeInstancesRequestWaiterParameters(instance.getInstanceId()), ec2Waiter.describeInstancesRequestWaiterHandler());
                }
            });

    instanceRepository.delete(instance);
}

@Override
public void process(Instance instance) {
    ec2UserClient.amazonEC2Async().rebootInstancesAsync(
            new RebootInstancesRequest().withInstanceIds(instance.getInstanceId()),
            new AsyncHandler<RebootInstancesRequest, RebootInstancesResult>() {
                @Override
                public void onError(Exception exception) {
                    log.warn("something went wrong starting the server {}",
                            exception.getLocalizedMessage());
                }

                @Override
                public void onSuccess(RebootInstancesRequest request,
                                      RebootInstancesResult result) {
                    ec2UserClient.amazonEC2Async().waiters()
                            .instanceRunning()
                            .runAsync(ec2Waiter.describeInstancesRequestWaiterParameters(instance.getInstanceId()), ec2Waiter.describeInstancesRequestWaiterHandler());

                }
            });
}

@Override
public void process(Instance instance) {
    ec2UserClient.amazonEC2Async().startInstancesAsync(
            new StartInstancesRequest().withInstanceIds(instance.getInstanceId()),
            new AsyncHandler<StartInstancesRequest, StartInstancesResult>() {
                @Override
                public void onError(Exception exception) {
                    log.warn("something went wrong starting the server {}",
                            exception.getLocalizedMessage());
                }

                @Override
                public void onSuccess(StartInstancesRequest request, StartInstancesResult result) {
                    onSuccessStart(instance);
                }
            });
}

private void runSearch(T type) {
    AmazonWebServiceRequest descRequest = buildRequest(type);
    AsyncHandler describeHandler = buildHandler(type);
    if (type instanceof Instance) {
        this.amazonEC2Client.describeInstancesAsync(
                (DescribeInstancesRequest) descRequest, describeHandler);
    } else if (type instanceof NatGateway) {
        this.amazonEC2Client.describeNatGatewaysAsync(
                (DescribeNatGatewaysRequest) descRequest, describeHandler);
    } else if (type instanceof Volume) {
        this.amazonEC2Client.describeVolumesAsync(
                (DescribeVolumesRequest) descRequest, describeHandler);
    } else {
        AWSTaskStatusChecker.this.taskManager.patchTaskToFailure(
                new IllegalArgumentException("Invalid type " + type));
    }
}

/**
 * Get the instances from AWS filtered by the instances Ids known to the local system.
 */
public void getRemoteInstances(EnumerationDeletionContext aws,
        AWSEnumerationDeletionSubStage next) {
    if (aws.localInstanceIds == null || aws.localInstanceIds.size() == 0) {
        logFine(() -> "No local records found. No states need to be fetched from the AWS"
                + " endpoint.");
        aws.subStage = next;
        deleteResourcesInLocalSystem(aws);
        return;
    }
    DescribeInstancesRequest request = new DescribeInstancesRequest();
    Filter runningInstanceFilter = getAWSNonTerminatedInstancesFilter();
    request.getFilters().add(runningInstanceFilter);
    // Get only the instances from the remote system for which a compute state exists in the
    // local system.
    logFine(() -> String.format("Fetching instance details for %d instances on the AWS"
                    + " endpoint.", aws.localInstanceIds.keySet().size()));
    request.getInstanceIds().addAll(new ArrayList<>(aws.localInstanceIds.keySet()));
    AsyncHandler<DescribeInstancesRequest, DescribeInstancesResult> resultHandler =
            new AWSEnumerationAsyncHandler(this, aws, next);
    aws.amazonEC2Client.describeInstancesAsync(request,
            resultHandler);
}

@Override
public void write(byte[] bytes) throws IOException {
  if (bytes == null || bytes.length == 0) {
    return;
  }

  final String msg = new String(bytes);

  if (bytes.length > maxMessageSizeInKB * 1024) {
    addWarn(format("Logging event '%s' exceeds the maximum size of %dkB",
        msg, maxMessageSizeInKB));
    return;
  }

  sqs.sendMessageAsync(new SendMessageRequest(queueUrl, msg),
      new AsyncHandler<SendMessageRequest, SendMessageResult>() {
        public void onError(Exception exception) {
          addWarn(format("Appender '%s' failed to send logging event '%s' to '%s'", getName(), msg, queueUrl), exception);
        }
        public void onSuccess(SendMessageRequest request, SendMessageResult result) {
          /** noop **/
        }
      });
}

protected void sendBatch(Map<String, Collection<MetricDatum>> datums) {
    for (final Map.Entry<String, Collection<MetricDatum>> e : datums.entrySet()) {
        for (final List<MetricDatum> batch : Lists.partition(Lists.newLinkedList(e.getValue()), BATCH_SIZE)) {
            cloudWatch.putMetricDataAsync(new PutMetricDataRequest().withNamespace(e.getKey())
                                                                    .withMetricData(batch),
                                          new AsyncHandler<PutMetricDataRequest, Void>() {

                                            @Override
                                            public void onError(Exception exception) {
                                                LOG.error("PutMetricData failed", exception);
                                                LOG.info("Requeueing metric data.");
                                                queuedDatums.putAll(e.getKey(), batch);
                                            }

                                            @Override
                                            public void onSuccess(PutMetricDataRequest request, Void result) {
                                                LOG.info("Successfully put " + request.getMetricData().size() +
                                                                        " datums for namespace " + request.getNamespace());
                                                LOG.debug("Request", request);
                                            }

            });
        }
    }
}

private void sendUsingAsyncClient(final PutRecordRequest putRecordRequest, Message<JsonObject> event) {

        if (retryCounter == 3) {
            sendError(event, "Failed sending message to Kinesis");
        }

        final Context ctx = vertx.currentContext();
        kinesisAsyncClient.putRecordAsync(putRecordRequest, new AsyncHandler<PutRecordRequest,PutRecordResult>() {
            public void onSuccess(PutRecordRequest request, final PutRecordResult recordResult) {
                ctx.runOnContext(v -> {
                    logger.debug("Sent message to Kinesis: " + recordResult.toString());
                    sendOK(event);
                });
            }
            public void onError(final java.lang.Exception iexc) {
                ctx.runOnContext(v -> {
                    retryCounter++;
                    kinesisAsyncClient = createClient();
                    logger.info("Failed sending message to Kinesis, retry: " + retryCounter + " ... ", iexc);
                    vertx.setTimer(500, timerID -> sendUsingAsyncClient(putRecordRequest, event));
                });
            }
        });
    }

@Override
public void process(Instance instance) {
    String keyName = ec2KeyPair.setName(instance);

    instance.setId(createUniqueId(keyName));
    instance.setUserId(authenticationService.getCurrentUser().getId());
    instance.setKeyName(keyName);
    instance.setKeyBlob(ec2KeyPair.create(instance));

    RunInstancesRequest runInstancesRequest = new RunInstancesRequest()
            .withInstanceType(instance.getInstanceType())
            .withImageId(instance.getImageId())
            .withMinCount(1)
            .withMaxCount(1)
            .withSecurityGroupIds(instance.getSecurityGroupId())
            .withKeyName(keyName)
            .withSubnetId(instance.getSubnetId())
            .withTagSpecifications(buildTags(instance));

    instanceRepository.save(instance);
    refreshClient(instance);

    ec2UserClient.amazonEC2Async().runInstancesAsync(runInstancesRequest,
            new AsyncHandler<RunInstancesRequest, RunInstancesResult>() {
                @Override
                public void onError(Exception exception) {
                    log.warn("something went wrong creating the server {}",
                            exception.getLocalizedMessage());
                }

                @Override
                public void onSuccess(RunInstancesRequest request, RunInstancesResult result) {
                    onSuccessCreate(result, instance);
                }
            });
}

private void validateAWSCredentials(final AWSInstanceContext aws) {
    if (aws.computeRequest.isMockRequest || AWSUtils.isAwsClientMock()) {
        aws.operation.complete();
        return;
    }

    // make a call to validate credentials
    aws.amazonEC2Client
            .describeAvailabilityZonesAsync(
                    new DescribeAvailabilityZonesRequest(),
                    new AsyncHandler<DescribeAvailabilityZonesRequest, DescribeAvailabilityZonesResult>() {
                        @Override
                        public void onError(Exception e) {
                            if (e instanceof AmazonServiceException) {
                                AmazonServiceException ase = (AmazonServiceException) e;
                                if (ase.getStatusCode() == STATUS_CODE_UNAUTHORIZED) {
                                    ServiceErrorResponse r = Utils.toServiceErrorResponse(e);
                                    r.statusCode = STATUS_CODE_UNAUTHORIZED;
                                    aws.operation.fail(e, r);
                                    return;
                                }
                            }

                            aws.operation.fail(e);
                        }

                        @Override
                        public void onSuccess(DescribeAvailabilityZonesRequest request,
                                DescribeAvailabilityZonesResult describeAvailabilityZonesResult) {
                            aws.operation.complete();
                        }
                    });
}

public static void validateCredentials(AmazonEC2AsyncClient ec2Client,
        AWSClientManager clientManager, AuthCredentialsServiceState credentials,
        ComputeEnumerateAdapterRequest context, Operation op, StatelessService service,
        Consumer<DescribeAvailabilityZonesResult> onSuccess, Consumer<Throwable> onFail) {

    if (clientManager.isEc2ClientInvalid(credentials, context.regionId)) {
        op.complete();
        return;
    }

    ec2Client.describeAvailabilityZonesAsync(new DescribeAvailabilityZonesRequest(),
            new AsyncHandler<DescribeAvailabilityZonesRequest, DescribeAvailabilityZonesResult>() {

                @Override
                public void onError(Exception e) {
                    if (e instanceof AmazonServiceException) {
                        AmazonServiceException ase = (AmazonServiceException) e;
                        if (ase.getStatusCode() == STATUS_CODE_UNAUTHORIZED) {
                            clientManager.markEc2ClientInvalid(service, credentials,
                                    context.regionId);
                            op.complete();
                            return;
                        }
                        onFail.accept(e);
                    }
                }

                @Override
                public void onSuccess(DescribeAvailabilityZonesRequest request,
                        DescribeAvailabilityZonesResult describeAvailabilityZonesResult) {
                    onSuccess.accept(describeAvailabilityZonesResult);
                }
            });
}

/**
 * Get or create a CloudWatch Client instance that will be used to get stats from AWS.
 * @param credentials
 *         The auth credentials to be used for the client creation
 * @param regionId
 *         The region of the AWS client
 * @param service
 *         The stateless service for which the operation is being performed.
 * @param isMock
 *         Indicates if this a mock request
 * @return
 */
public AmazonCloudWatchAsyncClient getOrCreateCloudWatchClient(
        AuthCredentialsServiceState credentials,
        String regionId, StatelessService service, boolean isMock,
        Consumer<Throwable> failConsumer) {
    if (this.awsClientType != AwsClientType.CLOUD_WATCH) {
        throw new UnsupportedOperationException(
                "This client manager supports only AWS " + this.awsClientType + " clients.");
    }
    String cacheKey = createCredentialRegionCacheKey(credentials, regionId);
    if (isCloudWatchClientInvalid(cacheKey)) {
        failConsumer.accept(
                new IllegalStateException("Invalid cloud watch client for key: " + cacheKey));
        return null;
    }
    AmazonCloudWatchAsyncClient amazonCloudWatchClient = null;
    try {
        amazonCloudWatchClient = this.cloudWatchClientCache.computeIfAbsent(cacheKey, key -> {
            AmazonCloudWatchAsyncClient client = AWSUtils.getStatsAsyncClient
                    (credentials, regionId, getExecutor(), isMock);
            client.describeAlarmsAsync(
                    new AsyncHandler<DescribeAlarmsRequest, DescribeAlarmsResult>() {
                        @Override
                        public void onError(Exception exception) {
                            markCloudWatchClientInvalid(service, cacheKey);
                        }

                        @Override
                        public void onSuccess(DescribeAlarmsRequest request,
                                DescribeAlarmsResult result) {
                            //noop
                        }
                    });
            return client;
        });
    } catch (Throwable e) {
        service.logSevere(e);
        failConsumer.accept(e);
    }
    return amazonCloudWatchClient;
}

/**
 * Initializes and saves a reference to the request object that is sent to AWS to get a page of
 * volumes. Also saves an instance to the async handler that will be used to handle the
 * responses received from AWS. It sets the nextToken value in the request object sent to AWS
 * for getting the next page of results from AWS.
 */
private void createAWSRequestAndAsyncHandler(EBSStorageEnumerationContext aws) {
    DescribeVolumesRequest request = new DescribeVolumesRequest();
    request.setMaxResults(getQueryPageSize());
    request.setNextToken(aws.nextToken);
    aws.describeVolumesRequest = request;
    AsyncHandler<DescribeVolumesRequest, DescribeVolumesResult> resultHandler = new AWSStorageEnumerationAsyncHandler(
            this, aws);
    aws.resultHandler = resultHandler;
}

/**
 * Deletes instances on the AWS endpoint for the set of instance Ids that are passed in.
 *
 * @param instanceIdsToDelete
 * @throws Throwable
 */
public static void deleteVMsUsingEC2Client(AmazonEC2AsyncClient client, VerificationHost host,
        List<String> instanceIdsToDelete) throws Throwable {
    TerminateInstancesRequest termRequest = new TerminateInstancesRequest(instanceIdsToDelete);
    AsyncHandler<TerminateInstancesRequest, TerminateInstancesResult> terminateHandler = new AWSTerminateHandlerAsync(
            host);
    client.terminateInstancesAsync(termRequest, terminateHandler);
    waitForInstancesToBeTerminated(client, host, instanceIdsToDelete);

}

/**
 * Stop instances on the AWS endpoint for the set of instance Ids that are passed in.
 *
 * @param client
 * @param host
 * @param instanceIdsToStop
 * @throws Throwable
 */
public static void stopVMsUsingEC2Client(AmazonEC2AsyncClient client, VerificationHost host,
        List<String> instanceIdsToStop) throws Throwable {
    StopInstancesRequest stopRequest = new StopInstancesRequest(instanceIdsToStop);
    AsyncHandler<StopInstancesRequest, StopInstancesResult> stopHandler = new AWSStopHandlerAsync(
            host);
    client.stopInstancesAsync(stopRequest, stopHandler);
    waitForInstancesToBeStopped(client, host, instanceIdsToStop);

}

@Override protected void doEnqueue(Callback<Void> callback) {
  future = get().putRecordAsync(message,
      new AsyncHandler<PutRecordRequest, PutRecordResult>() {
        @Override public void onError(Exception e) {
          callback.onError(e);
        }

        @Override
        public void onSuccess(PutRecordRequest request, PutRecordResult result) {
          callback.onSuccess(null);
        }
      });
  if (future.isCancelled()) throw new IllegalStateException("cancelled sending spans");
}

@Override protected void doEnqueue(Callback<Void> callback) {
  future = get().sendMessageAsync(message,
      new AsyncHandler<SendMessageRequest, SendMessageResult>() {
        @Override public void onError(Exception e) {
          callback.onError(e);
        }

        @Override
        public void onSuccess(SendMessageRequest request, SendMessageResult sendMessageResult) {
          callback.onSuccess(null);
        }
      });
  if (future.isCancelled()) throw new IllegalStateException("cancelled sending spans");
}

private <T extends Serializable> void sendMessageToAllRegionsAsync(final T body ) throws IOException {
    if ( snsAsync == null ) {
        logger.error( "SNS client is null, perhaps it failed to initialize successfully" );
        return;
    }

    final String stringBody = toString( body );

    final String topicArn = getWriteTopicArn();

    if ( logger.isTraceEnabled() ) {
        logger.trace( "Publishing Message...{} to arn: {}", stringBody, topicArn );
    }

    PublishRequest publishRequest = new PublishRequest( topicArn, stringBody );

    snsAsync.publishAsync( publishRequest, new AsyncHandler<PublishRequest, PublishResult>() {
        @Override
        public void onError( Exception e ) {
            logger.error( "Error publishing message... {}", e );
            logger.error(FAILED_TO_SEND_MESSAGE, stringBody, topicArn, e);
        }


        @Override
        public void onSuccess( PublishRequest request, PublishResult result ) {
            if ( logger.isTraceEnabled() ) {
                logger.trace( "Successfully published... messageID=[{}],  arn=[{}]", result.getMessageId(),
                    request.getTopicArn() );
            }
        }
    } );
}

@Test
@SuppressWarnings("unchecked")
public void testProducerErrorChannel() throws Exception {
    KinesisTestBinder binder = getBinder();

    final RuntimeException putRecordException = new RuntimeException("putRecordRequestEx");
    final AtomicReference<Object> sent = new AtomicReference<>();
    AmazonKinesisAsync amazonKinesisMock = mock(AmazonKinesisAsync.class);
    BDDMockito.given(amazonKinesisMock.putRecordAsync(any(PutRecordRequest.class), any(AsyncHandler.class)))
            .willAnswer((Answer<Future<PutRecordResult>>) invocation -> {
                PutRecordRequest request = invocation.getArgument(0);
                sent.set(request.getData());
                AsyncHandler<?, ?> handler = invocation.getArgument(1);
                handler.onError(putRecordException);
                return mock(Future.class);
            });

    new DirectFieldAccessor(binder.getBinder()).setPropertyValue("amazonKinesis", amazonKinesisMock);

    ExtendedProducerProperties<KinesisProducerProperties> producerProps = createProducerProperties();
    producerProps.setErrorChannelEnabled(true);
    DirectChannel moduleOutputChannel = createBindableChannel("output",
            createProducerBindingProperties(producerProps));
    Binding<MessageChannel> producerBinding = binder.bindProducer("ec.0", moduleOutputChannel, producerProps);

    ApplicationContext applicationContext = TestUtils.getPropertyValue(binder.getBinder(),
            "applicationContext", ApplicationContext.class);
    SubscribableChannel ec = applicationContext.getBean("ec.0.errors", SubscribableChannel.class);
    final AtomicReference<Message<?>> errorMessage = new AtomicReference<>();
    final CountDownLatch latch = new CountDownLatch(1);
    ec.subscribe(message -> {
        errorMessage.set(message);
        latch.countDown();
    });

    String messagePayload = "oops";
    moduleOutputChannel.send(new GenericMessage<>(messagePayload.getBytes()));

    assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
    assertThat(errorMessage.get()).isInstanceOf(ErrorMessage.class);
    assertThat(errorMessage.get().getPayload()).isInstanceOf(AwsRequestFailureException.class);
    AwsRequestFailureException exception = (AwsRequestFailureException) errorMessage.get().getPayload();
    assertThat(exception.getCause()).isSameAs(putRecordException);
    assertThat(((PutRecordRequest) exception.getRequest()).getData()).isSameAs(sent.get());
    producerBinding.unbind();
}

private void publishNotification(
    final String arn,
    final SNSMessage<?> message,
    final QualifiedName name,
    final String errorMessage,
    final String counterKey,
    final boolean retryOnLongMessage
) {
    this.notificationMetric.recordTime(message, Metrics.TimerNotificationsBeforePublishDelay.getMetricName());
    try {
        final AsyncHandler<PublishRequest, PublishResult> handler =
            new AsyncHandler<PublishRequest, PublishResult>() {
                @Override
                public void onError(final Exception exception) {
                    if (retryOnLongMessage && (exception instanceof InvalidParameterException
                        || exception instanceof InvalidParameterValueException)) {
                        log.error("SNS Publish message exceeded the size threshold", exception);
                        notificationMetric.counterIncrement(
                            Metrics.CounterSNSNotificationPublishMessageSizeExceeded.getMetricName());
                        final SNSMessage<Void> voidMessage = new SNSMessage<>(message.getId(),
                            message.getTimestamp(), message.getRequestId(), message.getType(), message.getName(),
                            null);
                        publishNotification(arn, voidMessage, name, errorMessage, counterKey, false);
                    } else {
                        notificationMetric.handleException(name, errorMessage, counterKey, message, exception);
                    }
                }

                @Override
                public void onSuccess(final PublishRequest request, final PublishResult publishResult) {
                    log.info("Successfully published message to topic {} with id {}",
                        arn, publishResult.getMessageId());
                    log.debug("Successfully published message {} to topic {} with id {}",
                        message, arn, publishResult.getMessageId());
                    notificationMetric.counterIncrement(counterKey);
                    notificationMetric.recordTime(message,
                        Metrics.TimerNotificationsPublishDelay.getMetricName());
                }
            };
        client.publishAsync(arn, mapper.writeValueAsString(message), handler);
    } catch (final Exception e) {
        notificationMetric.handleException(name, errorMessage, counterKey, message, e);
    }
}

/**
 * Gets EC2 statistics.
 *
 * @param statsData The context object for stats.
 * @param metricNames The metrics names to gather stats for.
 * @param isAggregateStats Indicates where we are interested in aggregate stats or not.
 */
private void getEC2Stats(AWSStatsDataHolder statsData, String[] metricNames,
        boolean isAggregateStats) {
    if (getAWSAsyncStatsClient(statsData) == null) {
        return;
    }
    Long collectionPeriod = Long.getLong(AWS_COLLECTION_PERIOD_SECONDS, DEFAULT_AWS_COLLECTION_PERIOD_SECONDS);
    for (String metricName : metricNames) {
        GetMetricStatisticsRequest metricRequest = new GetMetricStatisticsRequest();
        // get datapoint for the for the passed in time window.
        try {
            setRequestCollectionWindow(
                    TimeUnit.MINUTES.toMicros(MAX_METRIC_COLLECTION_WINDOW_IN_MINUTES),
                    statsData.statsRequest.lastCollectionTimeMicrosUtc,
                    collectionPeriod,
                    metricRequest);
        } catch (IllegalStateException e) {
            // no data to process. notify parent
            statsData.taskManager.finishTask();
            return;
        }
        metricRequest.setPeriod(collectionPeriod.intValue());
        metricRequest.setStatistics(Arrays.asList(STATISTICS));
        metricRequest.setNamespace(NAMESPACE);

        // Provide instance id dimension only if it is not aggregate stats.
        if (!isAggregateStats) {
            List<Dimension> dimensions = new ArrayList<>();
            Dimension dimension = new Dimension();
            dimension.setName(DIMENSION_INSTANCE_ID);
            String instanceId = statsData.computeDesc.id;
            dimension.setValue(instanceId);
            dimensions.add(dimension);
            metricRequest.setDimensions(dimensions);
        }

        metricRequest.setMetricName(metricName);

        logFine(() -> String.format("Retrieving %s metric from AWS", metricName));
        AsyncHandler<GetMetricStatisticsRequest, GetMetricStatisticsResult> resultHandler = new AWSStatsHandler(
                statsData, metricNames.length, isAggregateStats);
        statsData.statsClient.getMetricStatisticsAsync(metricRequest, resultHandler);
    }
}

private void getBillingStats(AWSStatsDataHolder statsData) {
    if (getAWSAsyncBillingClient(statsData) == null) {
        return;
    }
    Dimension dimension = new Dimension();
    dimension.setName(DIMENSION_CURRENCY);
    dimension.setValue(DIMENSION_CURRENCY_VALUE);
    GetMetricStatisticsRequest request = new GetMetricStatisticsRequest();
    // AWS pushes billing metrics every 4 hours. However the timeseries returned does not have
    // static time stamps associated with the data points. The timestamps range from
    // (currentTime - 4 hrs) and are spaced at 4 hrs.
    // Get all 14 days worth of estimated charges data by default when last collection time is not set.
    // Otherwise set the window to lastCollectionTime - 4 hrs.
    Long lastCollectionTimeForEstimatedCharges = null;
    Long collectionPeriod = Long.getLong(AWS_COLLECTION_PERIOD_SECONDS, DEFAULT_AWS_COLLECTION_PERIOD_SECONDS);
    if (statsData.statsRequest.lastCollectionTimeMicrosUtc != null) {
        lastCollectionTimeForEstimatedCharges =
                statsData.statsRequest.lastCollectionTimeMicrosUtc
                        - TimeUnit.HOURS.toMicros(COST_COLLECTION_PERIOD_IN_HOURS);

    }

    // defaulting to fetch 14 days of estimated charges data
    try {
        setRequestCollectionWindow(
                TimeUnit.DAYS.toMicros(COST_COLLECTION_WINDOW_IN_DAYS),
                lastCollectionTimeForEstimatedCharges,
                collectionPeriod,
                request);
    } catch (IllegalStateException e) {
        // no data to process. notify parent
        statsData.taskManager.finishTask();
        return;
    }

    request.setPeriod(COST_COLLECTION_PERIOD_IN_SECONDS);
    request.setStatistics(Arrays.asList(STATISTICS));
    request.setNamespace(BILLING_NAMESPACE);
    request.setDimensions(Collections.singletonList(dimension));
    request.setMetricName(AWSConstants.ESTIMATED_CHARGES);

    logFine(() -> String.format("Retrieving %s metric from AWS",
            AWSConstants.ESTIMATED_CHARGES));
    AsyncHandler<GetMetricStatisticsRequest, GetMetricStatisticsResult> resultHandler = new AWSBillingStatsHandler(
            statsData, lastCollectionTimeForEstimatedCharges);
    statsData.billingClient.getMetricStatisticsAsync(request, resultHandler);
}

/**
 * Creates an ebs volume.
 */
public void createVolume(CreateVolumeRequest req,
        AsyncHandler<CreateVolumeRequest, CreateVolumeResult> creationHandler) {
    this.client.createVolumeAsync(req, creationHandler);
}

/**
 * Deletes an ebs volume.
 */
public void deleteVolume(DeleteVolumeRequest req,
        AsyncHandler<DeleteVolumeRequest, DeleteVolumeResult> deletionHandler) {
    this.client.deleteVolumeAsync(req, deletionHandler);
}