AWSAppAutoScalingClient implements AppAutoScalingClient { @Override public Completable deleteScalableTarget(String jobId) { DeregisterScalableTargetRequest deRegisterRequest = new DeregisterScalableTargetRequest(); deRegisterRequest.setResourceId( AWSAppAutoScalingUtil.buildGatewayResourceId(jobId, awsAppScalingConfig.getAWSGatewayEndpointPrefix(), awsAppScalingConfig.getRegion(), awsAppScalingConfig.getAWSGatewayEndpointTargetStage())); deRegisterRequest.setServiceNamespace(SERVICE_NAMESPACE); deRegisterRequest.setScalableDimension(SCALABLE_DIMENSION); return RetryWrapper.wrapWithExponentialRetry(String.format("deleteScalableTarget for job %s", jobId), Observable.create(emitter -> awsAppAutoScalingClientAsync.deregisterScalableTargetAsync(deRegisterRequest, new AsyncHandler<DeregisterScalableTargetRequest, DeregisterScalableTargetResult>() { @Override public void onError(Exception exception) { if (exception instanceof ObjectNotFoundException) { logger.info("Scalable target does not exist anymore for job {}", jobId); emitter.onCompleted(); } else { logger.error("Deregister scalable target exception {} - {}", jobId, exception.getMessage()); awsAppAutoScalingMetrics.registerAwsDeleteTargetError(exception); emitter.onError(exception); } } @Override public void onSuccess(DeregisterScalableTargetRequest request, DeregisterScalableTargetResult deregisterScalableTargetResult) { int httpStatusCode = deregisterScalableTargetResult.getSdkHttpMetadata().getHttpStatusCode(); logger.info("De-registered scalable target for {}, status {}", jobId, httpStatusCode); awsAppAutoScalingMetrics.registerAwsDeleteTargetSuccess(); emitter.onCompleted(); } }), Emitter.BackpressureMode.NONE)).toCompletable(); } @Inject  AWSAppAutoScalingClient(AWSCredentialsProvider awsCredentialsProvider,
                                   AWSAppScalingConfig awsAppScalingConfig,
                                   Registry registry); @VisibleForTesting  AWSAppAutoScalingClient(AWSApplicationAutoScalingAsync awsAppAutoScalingClientAsync, AWSAppScalingConfig awsAppScalingConfig, Registry registry); @Override Completable createScalableTarget(String jobId, int minCapacity, int maxCapacity); @Override Observable<AutoScalableTarget> getScalableTargetsForJob(String jobId); @Override Observable<String> createOrUpdateScalingPolicy(String policyRefId, String jobId, PolicyConfiguration policyConfiguration); @Override Completable deleteScalableTarget(String jobId); @Override Completable deleteScalingPolicy(String policyRefId, String jobId); final static String SERVICE_NAMESPACE; final static String SCALABLE_DIMENSION; }
@Test public void deleteScalableTargetIsIdempotent() { String jobId = UUID.randomUUID().toString(); String policyId = UUID.randomUUID().toString(); AWSApplicationAutoScalingAsync clientAsync = mock(AWSApplicationAutoScalingAsync.class); AWSAppScalingConfig config = mock(AWSAppScalingConfig.class); AWSAppAutoScalingClient autoScalingClient = new AWSAppAutoScalingClient(clientAsync, config, new NoopRegistry()); AtomicBoolean isDeleted = new AtomicBoolean(false); when(clientAsync.deregisterScalableTargetAsync(any(), any())).thenAnswer(invocation -> { DeregisterScalableTargetRequest request = invocation.getArgument(0); AsyncHandler<DeregisterScalableTargetRequest, DeregisterScalableTargetResult> handler = invocation.getArgument(1); if (isDeleted.get()) { ObjectNotFoundException notFoundException = new ObjectNotFoundException(policyId + " does not exist"); handler.onError(notFoundException); return Future.failed(notFoundException); } DeregisterScalableTargetResult resultSuccess = new DeregisterScalableTargetResult(); HttpResponse successResponse = new HttpResponse(null, null); successResponse.setStatusCode(200); resultSuccess.setSdkHttpMetadata(SdkHttpMetadata.from(successResponse)); isDeleted.set(true); handler.onSuccess(request, resultSuccess); return Future.successful(resultSuccess); }); AssertableSubscriber<Void> firstCall = autoScalingClient.deleteScalableTarget(jobId).test(); firstCall.awaitTerminalEvent(2, TimeUnit.SECONDS); firstCall.assertNoErrors(); firstCall.assertCompleted(); verify(clientAsync, times(1)).deregisterScalableTargetAsync(any(), any()); AssertableSubscriber<Void> secondCall = autoScalingClient.deleteScalableTarget(jobId).test(); secondCall.awaitTerminalEvent(2, TimeUnit.SECONDS); secondCall.assertNoErrors(); secondCall.assertCompleted(); verify(clientAsync, times(2)).deregisterScalableTargetAsync(any(), any()); }