WorkflowServiceImpl extends AbstractIndexProducer implements WorkflowService, JobProducer, ManagedService { @Override public synchronized void cleanupWorkflowInstances(int buffer, WorkflowState state) throws UnauthorizedException, WorkflowDatabaseException { logger.info("Start cleaning up workflow instances older than {} days with status '{}'", buffer, state); int instancesCleaned = 0; int cleaningFailed = 0; WorkflowQuery query = new WorkflowQuery().withState(state).withDateBefore(DateUtils.addDays(new Date(), -buffer)) .withCount(Integer.MAX_VALUE); for (WorkflowInstance workflowInstance : getWorkflowInstances(query).getItems()) { try { remove(workflowInstance.getId()); instancesCleaned++; } catch (WorkflowDatabaseException | UnauthorizedException e) { throw e; } catch (NotFoundException e) { logger.debug("Workflow instance '{}' could not be removed", workflowInstance.getId(), e); } catch (WorkflowParsingException | WorkflowStateException e) { logger.warn("Workflow instance '{}' could not be removed", workflowInstance.getId(), e); cleaningFailed++; } } if (instancesCleaned == 0 && cleaningFailed == 0) { logger.info("No workflow instances found to clean up"); return; } if (instancesCleaned > 0) logger.info("Cleaned up '%d' workflow instances", instancesCleaned); if (cleaningFailed > 0) { logger.warn("Cleaning failed for '%d' workflow instances", cleaningFailed); throw new WorkflowDatabaseException("Unable to clean all workflow instances, see logs!"); } } WorkflowServiceImpl(); @Activate void activate(ComponentContext componentContext); @Override @Deactivate void deactivate(); @Override void addWorkflowListener(WorkflowListener listener); @Override void removeWorkflowListener(WorkflowListener listener); @Override List<WorkflowDefinition> listAvailableWorkflowDefinitions(); boolean isRunnable(WorkflowDefinition workflowDefinition); Set<HandlerRegistration> getRegisteredHandlers(); @Override WorkflowInstanceImpl getWorkflowById(long id); @Override WorkflowInstance start(WorkflowDefinition workflowDefinition, MediaPackage mediaPackage); @Override WorkflowInstance start(WorkflowDefinition workflowDefinition, MediaPackage mediaPackage,
          Map<String, String> properties); @Override WorkflowInstance start(WorkflowDefinition workflowDefinition, MediaPackage sourceMediaPackage,
          Long parentWorkflowId, Map<String, String> originalProperties); @Override WorkflowDefinition getWorkflowDefinitionById(String id); @Override WorkflowInstance stop(long workflowInstanceId); @Override void remove(long workflowInstanceId); @Override void remove(long workflowInstanceId, boolean force); @Override WorkflowInstance suspend(long workflowInstanceId); @Override WorkflowInstance resume(long id); @Override WorkflowInstance resume(long workflowInstanceId, Map<String, String> properties); @Override void update(final WorkflowInstance workflowInstance); @Override long countWorkflowInstances(); @Override long countWorkflowInstances(WorkflowState state, String operation); @Override WorkflowStatistics getStatistics(); @Override WorkflowSet getWorkflowInstances(WorkflowQuery query); @Override WorkflowSet getWorkflowInstancesForAdministrativeRead(WorkflowQuery query); @Override boolean isReadyToAcceptJobs(String operation); @Override boolean isReadyToAccept(Job job); @Override synchronized void acceptJob(Job job); @Override long countJobs(Status status); ServiceRegistry getServiceRegistry(); @Reference(name = "security-service") void setSecurityService(SecurityService securityService); @Reference(name = "authorization") void setAuthorizationService(AuthorizationService authorizationService); @Reference(name = "user-directory") void setUserDirectoryService(UserDirectoryService userDirectoryService); @Reference(name = "orgDirectory") void setOrganizationDirectoryService(OrganizationDirectoryService organizationDirectory); @Reference(name = "series") void setSeriesService(SeriesService seriesService); @Reference(name = "assetManager") void setAssetManager(AssetManager assetManager); @Reference(name = "message-broker-receiver") void setMessageReceiver(MessageReceiver messageReceiver); @Reference(name = "message-broker-sender") void setMessageSender(MessageSender messageSender); @Override String getJobType(); @Override @SuppressWarnings("rawtypes") void updated(Dictionary properties); @Override synchronized void cleanupWorkflowInstances(int buffer, WorkflowState state); @Override Map<String, Map<String, String>> getWorkflowStateMappings(); @Override void repopulate(final String indexName); @Override MessageReceiver getMessageReceiver(); @Override Service getService(); @Override String getClassName(); @Override MessageSender getMessageSender(); @Override SecurityService getSecurityService(); @Override String getSystemUserName(); static final String STATS_COLLECT_CONFIG_KEY; static final Boolean DEFAULT_STATS_COLLECT_CONFIG; static final String ERROR_RESOLUTION_HANDLER_ID; }
@Test public void testCleanupWorkflowInstances() throws Exception { WorkflowInstance wi1 = startAndWait(workingDefinition, mediapackage1, WorkflowState.SUCCEEDED); startAndWait(workingDefinition, mediapackage2, WorkflowState.SUCCEEDED); wi1 = service.getWorkflowById(wi1.getId()); service.cleanupWorkflowInstances(0, WorkflowState.FAILED); assertEquals(2, service.getWorkflowInstances(new WorkflowQuery()).size()); service.cleanupWorkflowInstances(0, WorkflowState.SUCCEEDED); assertEquals(0, service.getWorkflowInstances(new WorkflowQuery()).size()); for (WorkflowOperationInstance op : wi1.getOperations()) { assertEquals(0, serviceRegistry.getChildJobs(op.getId()).size()); } }