@Test public void cancelTooOldTerminationFlowForResourceTest() throws TransactionService.TransactionExecutionException { Set<FlowLogIdWithTypeAndTimestamp> flowLogs = new LinkedHashSet<>(); FlowLogIdWithTypeAndTimestamp flowLog2 = mock(FlowLogIdWithTypeAndTimestamp.class); Class flow2Class = Class.class; when(flowLog2.getFlowType()).thenReturn(flow2Class); flowLogs.add(flowLog2); FlowLogIdWithTypeAndTimestamp flowLog1 = mock(FlowLogIdWithTypeAndTimestamp.class); Class termFlowClass = TerminationFlowConfig.class; when(flowLog1.getFlowType()).thenReturn(termFlowClass); when(flowLog1.getCreated()).thenReturn(9000L); when(flowLog1.getFlowId()).thenReturn("flow1"); flowLogs.add(flowLog1); when(flowLogRepository.findAllRunningFlowLogByResourceId(eq(1L))).thenReturn(flowLogs); FlowLog realFlowLog1 = mock(FlowLog.class); when(realFlowLog1.getId()).thenReturn(10L); when(flowLogRepository.findFirstByFlowIdOrderByCreatedDesc(eq("flow1"))).thenReturn(Optional.of(realFlowLog1)); when(applicationFlowInformation.getTerminationFlow()).thenReturn(Collections.singletonList(TerminationFlowConfig.class)); when(transactionService.required(any(Supplier.class))).thenAnswer(invocation -> ((Supplier) invocation.getArguments()[0]).get()); when(nodeConfig.getId()).thenReturn("node1"); underTest.cancelTooOldTerminationFlowForResource(1L, 10000L); verify(flowLogRepository).finalizeByFlowId(eq("flow1")); verify(flowLogRepository, times(0)).finalizeByFlowId(eq("flow2")); verify(flowLogRepository).updateLastLogStatusInFlow(eq(10L), eq(StateStatus.SUCCESSFUL)); }
public void cancelTooOldTerminationFlowForResource(Long resourceId, long olderThan) { Set<FlowLogIdWithTypeAndTimestamp> allRunningFlowIdsByResourceId = flowLogRepository.findAllRunningFlowLogByResourceId(resourceId); allRunningFlowIdsByResourceId.stream() .filter(flowLog -> applicationFlowInformation.getTerminationFlow().stream() .map(Class::getName) .anyMatch(terminationFlowClassName -> terminationFlowClassName.equals(flowLog.getFlowType().getName()))) .filter(flowlog -> flowlog.getCreated() < olderThan) .findFirst().ifPresent(flowLog -> { try { LOGGER.info("Cancel flow [{}] for resource [{}] because it's too old", flowLog.getFlowId(), resourceId); cancel(resourceId, flowLog.getFlowId()); } catch (TransactionExecutionException e) { LOGGER.error("Can't cancel termination flow: {}", flowLog.getFlowId(), e); } }); }
FlowLogDBService implements FlowLogService { public void cancelTooOldTerminationFlowForResource(Long resourceId, long olderThan) { Set<FlowLogIdWithTypeAndTimestamp> allRunningFlowIdsByResourceId = flowLogRepository.findAllRunningFlowLogByResourceId(resourceId); allRunningFlowIdsByResourceId.stream() .filter(flowLog -> applicationFlowInformation.getTerminationFlow().stream() .map(Class::getName) .anyMatch(terminationFlowClassName -> terminationFlowClassName.equals(flowLog.getFlowType().getName()))) .filter(flowlog -> flowlog.getCreated() < olderThan) .findFirst().ifPresent(flowLog -> { try { LOGGER.info("Cancel flow [{}] for resource [{}] because it's too old", flowLog.getFlowId(), resourceId); cancel(resourceId, flowLog.getFlowId()); } catch (TransactionExecutionException e) { LOGGER.error("Can't cancel termination flow: {}", flowLog.getFlowId(), e); } }); } }
FlowLogDBService implements FlowLogService { public void cancelTooOldTerminationFlowForResource(Long resourceId, long olderThan) { Set<FlowLogIdWithTypeAndTimestamp> allRunningFlowIdsByResourceId = flowLogRepository.findAllRunningFlowLogByResourceId(resourceId); allRunningFlowIdsByResourceId.stream() .filter(flowLog -> applicationFlowInformation.getTerminationFlow().stream() .map(Class::getName) .anyMatch(terminationFlowClassName -> terminationFlowClassName.equals(flowLog.getFlowType().getName()))) .filter(flowlog -> flowlog.getCreated() < olderThan) .findFirst().ifPresent(flowLog -> { try { LOGGER.info("Cancel flow [{}] for resource [{}] because it's too old", flowLog.getFlowId(), resourceId); cancel(resourceId, flowLog.getFlowId()); } catch (TransactionExecutionException e) { LOGGER.error("Can't cancel termination flow: {}", flowLog.getFlowId(), e); } }); }  }
FlowLogDBService implements FlowLogService { public void cancelTooOldTerminationFlowForResource(Long resourceId, long olderThan) { Set<FlowLogIdWithTypeAndTimestamp> allRunningFlowIdsByResourceId = flowLogRepository.findAllRunningFlowLogByResourceId(resourceId); allRunningFlowIdsByResourceId.stream() .filter(flowLog -> applicationFlowInformation.getTerminationFlow().stream() .map(Class::getName) .anyMatch(terminationFlowClassName -> terminationFlowClassName.equals(flowLog.getFlowType().getName()))) .filter(flowlog -> flowlog.getCreated() < olderThan) .findFirst().ifPresent(flowLog -> { try { LOGGER.info("Cancel flow [{}] for resource [{}] because it's too old", flowLog.getFlowId(), resourceId); cancel(resourceId, flowLog.getFlowId()); } catch (TransactionExecutionException e) { LOGGER.error("Can't cancel termination flow: {}", flowLog.getFlowId(), e); } }); }  FlowLog save(FlowParameters flowParameters, String flowChanId, String key, Payload payload, Map<Object, Object> variables, Class<?> flowType,
            FlowState currentState); String getSerializedString(Object object); @Override Iterable<FlowLog> saveAll(Iterable<FlowLog> flowLogs); FlowLog close(Long stackId, String flowId); FlowLog cancel(Long stackId, String flowId); FlowLog terminate(Long stackId, String flowId); void saveChain(String flowChainId, String parentFlowChainId, Queue<Selectable> chain, String flowTriggerUserCrn); void updateLastFlowLogStatus(FlowLog lastFlowLog, boolean failureEvent); void cancelTooOldTerminationFlowForResource(Long resourceId, long olderThan); boolean isOtherNonTerminationFlowRunning(Long resourceId); boolean isOtherFlowRunning(Long resourceId); boolean repeatedFlowState(FlowLog lastFlowLog, String event); void updateLastFlowLogPayload(FlowLog lastFlowLog, Payload payload, Map<Object, Object> variables); Optional<FlowLog> getLastFlowLog(String flowId); @Override Set<String> findAllRunningNonTerminationFlowIdsByStackId(Long resourceId); @Override Optional<FlowLog> findFirstByFlowIdOrderByCreatedDesc(String flowId); @Override Optional<FlowChainLog> findFirstByFlowChainIdOrderByCreatedDesc(String flowChainId); @Override List<Object[]> findAllPending(); @Override Set<FlowLog> findAllUnassigned(); @Override Set<FlowLog> findAllByCloudbreakNodeId(String cloudbreakNodeId); List<FlowLog> findAllByResourceIdOrderByCreatedDesc(Long id); List<FlowLog> findAllByResourceIdAndFinalizedIsFalseOrderByCreatedDesc(Long id); @Override int purgeFinalizedFlowLogs(); List<FlowLog> findAllByFlowIdOrderByCreatedDesc(String flowId); Long getResourceIdByCrnOrName(String resource); FlowLog getLastFlowLogByResourceCrnOrName(String resource); List<FlowLog> getFlowLogsByResourceCrnOrName(String resource); Set<String> getFlowIdsByChainIds(Set<String> flowChainIds); List<FlowLog> getFlowLogsByFlowIdsCreatedDesc(Set<String> flowIds); Boolean hasPendingFlowEvent(List<FlowLog> flowLogs); Predicate<FlowLog> pendingFlowLogPredicate(); }
FlowLogDBService implements FlowLogService { public void cancelTooOldTerminationFlowForResource(Long resourceId, long olderThan) { Set<FlowLogIdWithTypeAndTimestamp> allRunningFlowIdsByResourceId = flowLogRepository.findAllRunningFlowLogByResourceId(resourceId); allRunningFlowIdsByResourceId.stream() .filter(flowLog -> applicationFlowInformation.getTerminationFlow().stream() .map(Class::getName) .anyMatch(terminationFlowClassName -> terminationFlowClassName.equals(flowLog.getFlowType().getName()))) .filter(flowlog -> flowlog.getCreated() < olderThan) .findFirst().ifPresent(flowLog -> { try { LOGGER.info("Cancel flow [{}] for resource [{}] because it's too old", flowLog.getFlowId(), resourceId); cancel(resourceId, flowLog.getFlowId()); } catch (TransactionExecutionException e) { LOGGER.error("Can't cancel termination flow: {}", flowLog.getFlowId(), e); } }); }  FlowLog save(FlowParameters flowParameters, String flowChanId, String key, Payload payload, Map<Object, Object> variables, Class<?> flowType,
            FlowState currentState); String getSerializedString(Object object); @Override Iterable<FlowLog> saveAll(Iterable<FlowLog> flowLogs); FlowLog close(Long stackId, String flowId); FlowLog cancel(Long stackId, String flowId); FlowLog terminate(Long stackId, String flowId); void saveChain(String flowChainId, String parentFlowChainId, Queue<Selectable> chain, String flowTriggerUserCrn); void updateLastFlowLogStatus(FlowLog lastFlowLog, boolean failureEvent); void cancelTooOldTerminationFlowForResource(Long resourceId, long olderThan); boolean isOtherNonTerminationFlowRunning(Long resourceId); boolean isOtherFlowRunning(Long resourceId); boolean repeatedFlowState(FlowLog lastFlowLog, String event); void updateLastFlowLogPayload(FlowLog lastFlowLog, Payload payload, Map<Object, Object> variables); Optional<FlowLog> getLastFlowLog(String flowId); @Override Set<String> findAllRunningNonTerminationFlowIdsByStackId(Long resourceId); @Override Optional<FlowLog> findFirstByFlowIdOrderByCreatedDesc(String flowId); @Override Optional<FlowChainLog> findFirstByFlowChainIdOrderByCreatedDesc(String flowChainId); @Override List<Object[]> findAllPending(); @Override Set<FlowLog> findAllUnassigned(); @Override Set<FlowLog> findAllByCloudbreakNodeId(String cloudbreakNodeId); List<FlowLog> findAllByResourceIdOrderByCreatedDesc(Long id); List<FlowLog> findAllByResourceIdAndFinalizedIsFalseOrderByCreatedDesc(Long id); @Override int purgeFinalizedFlowLogs(); List<FlowLog> findAllByFlowIdOrderByCreatedDesc(String flowId); Long getResourceIdByCrnOrName(String resource); FlowLog getLastFlowLogByResourceCrnOrName(String resource); List<FlowLog> getFlowLogsByResourceCrnOrName(String resource); Set<String> getFlowIdsByChainIds(Set<String> flowChainIds); List<FlowLog> getFlowLogsByFlowIdsCreatedDesc(Set<String> flowIds); Boolean hasPendingFlowEvent(List<FlowLog> flowLogs); Predicate<FlowLog> pendingFlowLogPredicate();  }