class ProvisionerService {

    public void waitCloudbreakClusterDeletion(Long id, PollingConfig pollingConfig) {
        SdxCluster sdxCluster = sdxService.getById(id);
        AtomicInteger deleteFailedCount = new AtomicInteger(1);
        Polling.waitPeriodly(pollingConfig.getSleepTime(), pollingConfig.getSleepTimeUnit())
                .stopIfException(pollingConfig.getStopPollingIfExceptionOccured())
                .stopAfterDelay(pollingConfig.getDuration(), pollingConfig.getDurationTimeUnit())
                .run(() -> {
                    LOGGER.info("Deletion polling cloudbreak for stack status: '{}' in '{}' env", sdxCluster.getClusterName(), sdxCluster.getEnvName());
                    try {
                        StackV4Response stackV4Response = ThreadBasedUserCrnProvider.doAsInternalActor(() -> stackV4Endpoint
                                .get(0L, sdxCluster.getClusterName(), Collections.emptySet(), sdxCluster.getAccountId()));
                        LOGGER.info("Stack status of SDX {} by response from cloudbreak: {}", sdxCluster.getClusterName(),
                                stackV4Response.getStatus().name());
                        LOGGER.debug("Response from cloudbreak: {}", JsonUtil.writeValueAsString(stackV4Response));
                        ClusterV4Response cluster = stackV4Response.getCluster();
                        if (cluster != null) {
                            if (StatusKind.PROGRESS.equals(cluster.getStatus().getStatusKind())) {
                                return AttemptResults.justContinue();
                            }
                            if (Status.DELETE_FAILED.equals(cluster.getStatus())) {
                                // it's a hack, until we implement a nice non-async terminate which can return with flowid..
                                // if it is implemented, please remove this
                                if (deleteFailedCount.getAndIncrement() >= DELETE_FAILED_RETRY_COUNT) {
                                    LOGGER.error("Cluster deletion failed '" + sdxCluster.getClusterName() + "', "
                                            + stackV4Response.getCluster().getStatusReason());
                                    return AttemptResults.breakFor(
                                            "Data Lake deletion failed '" + sdxCluster.getClusterName() + "', "
                                                    + stackV4Response.getCluster().getStatusReason()
                                    );
                                } else {
                                    return AttemptResults.justContinue();
                                }
                            }
                        }
                        if (Status.DELETE_FAILED.equals(stackV4Response.getStatus())) {
                            // it's a hack, until we implement a nice non-async terminate which can return with flowid..
                            // if it is implemented, please remove this
                            if (deleteFailedCount.getAndIncrement() >= DELETE_FAILED_RETRY_COUNT) {
                                LOGGER.error("Stack deletion failed '" + sdxCluster.getClusterName() + "', " + stackV4Response.getStatusReason());
                                return AttemptResults.breakFor(
                                        "Data Lake deletion failed '" + sdxCluster.getClusterName() + "', " + stackV4Response.getStatusReason()
                                );
                            } else {
                                return AttemptResults.justContinue();
                            }
                        } else {
                            return AttemptResults.justContinue();
                        }
                    } catch (NotFoundException e) {
                        return AttemptResults.finishWith(null);
                    }
                });
        sdxStatusService.setStatusForDatalakeAndNotify(DatalakeStatusEnum.STACK_DELETED, "Datalake stack deleted", sdxCluster);
    }

}