class OperationRetryService {

    public FlowIdentifier retry(Long stackId) {
        if (isFlowPending(stackId)) {
            LOGGER.info("Retry cannot be performed, because there is already an active flow. stackId: {}", stackId);
            throw new BadRequestException("Retry cannot be performed, because there is already an active flow.");
        }

        List<FlowLog> flowLogs = flowLogRepository.findAllByResourceIdOrderByCreatedDesc(stackId, PageRequest.of(0, LAST_FIFTY_FLOWLOGS));
        List<RetryableFlow> retryableFlows = getRetryableFlows(flowLogs);
        if (CollectionUtils.isEmpty(retryableFlows)) {
            LOGGER.info("Retry cannot be performed. The last flow did not fail or not retryable. stackId: {}", stackId);
            throw new BadRequestException("Retry cannot be performed. The last flow did not fail or not retryable.");
        }

        String name = retryableFlows.get(0).getName();
        eventService.fireCloudbreakEvent(stackId, Status.UPDATE_IN_PROGRESS.name(), STACK_RETRY_FLOW_START, List.of(name));

        Optional<FlowLog> failedFlowLog = getMostRecentFailedLog(flowLogs);
        Optional<FlowLog> lastSuccessfulStateFlowLog = failedFlowLog.map(log -> getLastSuccessfulStateLog(log.getCurrentState(), flowLogs));
        if (lastSuccessfulStateFlowLog.isPresent()) {
            flow2Handler.restartFlow(lastSuccessfulStateFlowLog.get());
            if (lastSuccessfulStateFlowLog.get().getFlowChainId() != null) {
                return new FlowIdentifier(FlowType.FLOW_CHAIN, lastSuccessfulStateFlowLog.get().getFlowChainId());
            } else {
                return new FlowIdentifier(FlowType.FLOW, lastSuccessfulStateFlowLog.get().getFlowId());
            }
        } else {
            LOGGER.info("Cannot restart previous flow because there is no successful state in the flow. stackId: {}", stackId);
            throw new BadRequestException("Cannot restart previous flow because there is no successful state in the flow.");
        }
    }

}