@Test void stuckTasksNotInProcessingStateWillBeResumed() { when(taskDao.markAsSubmitted(any(), anyLong(), any())).thenReturn(true); StuckTask task = new ITaskDao.StuckTask() .setType("TEST") .setStatus(TaskStatus.SUBMITTED.name()) .setVersionId(new TaskVersionId(UuidUtils.generatePrefixCombUuid(), 0)); AtomicInteger numResumed = new AtomicInteger(); service.handleStuckTask(task, numResumed, null, null); verify(tasksExecutionTriggerer).trigger(argThat( baseTask -> 1 == baseTask.getVersion() && task.getVersionId().getId().equals(baseTask.getId()) )); verify(taskDao).markAsSubmitted( task.getVersionId().getId(), task.getVersionId().getVersion(), now.plusMinutes(10) ); }
protected void handleStuckTask(ITaskDao.StuckTask task, AtomicInteger resumedCount, AtomicInteger errorCount, AtomicInteger failedCount) { ITaskProcessingPolicy.StuckTaskResolutionStrategy taskResolutionStrategy = null; ITaskHandler taskHandler = taskHandlerRegistry.getTaskHandler(task); String bucketId = null; ITaskProcessingPolicy taskProcessingPolicy = null; if (taskHandler == null) { log.error("No task handler found for task " + LogUtils.asParameter(task.getVersionId()) + "."); } else { taskProcessingPolicy = taskHandler.getProcessingPolicy(task); if (taskProcessingPolicy == null) { log.error("No processing policy found for task " + LogUtils.asParameter(task.getVersionId()) + "."); } else { taskResolutionStrategy = taskProcessingPolicy.getStuckTaskResolutionStrategy(task); bucketId = taskProcessingPolicy.getProcessingBucket(task); } } if (!TaskStatus.PROCESSING.name().equals(task.getStatus())) { retryTask(taskProcessingPolicy, bucketId, task, resumedCount); return; } if (taskResolutionStrategy == null) { log.error("No processing policy found for task " + LogUtils.asParameter(task.getVersionId()) + "."); markTaskAsError(bucketId, task, errorCount); return; } switch (taskResolutionStrategy) { case RETRY: retryTask(taskProcessingPolicy, bucketId, task, resumedCount); break; case MARK_AS_ERROR: markTaskAsError(bucketId, task, errorCount); break; case MARK_AS_FAILED: if (!taskDao.setStatus(task.getVersionId().getId(), TaskStatus.FAILED, task.getVersionId().getVersion())) { meterHelper.registerFailedStatusChange(task.getType(), task.getStatus(), TaskStatus.FAILED); } else { failedCount.getAndIncrement(); String taskType = task.getType(); meterHelper.registerStuckTaskMarkedAsFailed(taskType); meterHelper.registerTaskMarkedAsFailed(bucketId, taskType); } break; case IGNORE: meterHelper.registerStuckTaskAsIgnored(task.getType()); break; default: throw new UnsupportedOperationException("Resolution strategy " + taskResolutionStrategy + " is not supported."); } }
TasksResumer implements ITasksResumer, GracefulShutdownStrategy { protected void handleStuckTask(ITaskDao.StuckTask task, AtomicInteger resumedCount, AtomicInteger errorCount, AtomicInteger failedCount) { ITaskProcessingPolicy.StuckTaskResolutionStrategy taskResolutionStrategy = null; ITaskHandler taskHandler = taskHandlerRegistry.getTaskHandler(task); String bucketId = null; ITaskProcessingPolicy taskProcessingPolicy = null; if (taskHandler == null) { log.error("No task handler found for task " + LogUtils.asParameter(task.getVersionId()) + "."); } else { taskProcessingPolicy = taskHandler.getProcessingPolicy(task); if (taskProcessingPolicy == null) { log.error("No processing policy found for task " + LogUtils.asParameter(task.getVersionId()) + "."); } else { taskResolutionStrategy = taskProcessingPolicy.getStuckTaskResolutionStrategy(task); bucketId = taskProcessingPolicy.getProcessingBucket(task); } } if (!TaskStatus.PROCESSING.name().equals(task.getStatus())) { retryTask(taskProcessingPolicy, bucketId, task, resumedCount); return; } if (taskResolutionStrategy == null) { log.error("No processing policy found for task " + LogUtils.asParameter(task.getVersionId()) + "."); markTaskAsError(bucketId, task, errorCount); return; } switch (taskResolutionStrategy) { case RETRY: retryTask(taskProcessingPolicy, bucketId, task, resumedCount); break; case MARK_AS_ERROR: markTaskAsError(bucketId, task, errorCount); break; case MARK_AS_FAILED: if (!taskDao.setStatus(task.getVersionId().getId(), TaskStatus.FAILED, task.getVersionId().getVersion())) { meterHelper.registerFailedStatusChange(task.getType(), task.getStatus(), TaskStatus.FAILED); } else { failedCount.getAndIncrement(); String taskType = task.getType(); meterHelper.registerStuckTaskMarkedAsFailed(taskType); meterHelper.registerTaskMarkedAsFailed(bucketId, taskType); } break; case IGNORE: meterHelper.registerStuckTaskAsIgnored(task.getType()); break; default: throw new UnsupportedOperationException("Resolution strategy " + taskResolutionStrategy + " is not supported."); } } }
TasksResumer implements ITasksResumer, GracefulShutdownStrategy { protected void handleStuckTask(ITaskDao.StuckTask task, AtomicInteger resumedCount, AtomicInteger errorCount, AtomicInteger failedCount) { ITaskProcessingPolicy.StuckTaskResolutionStrategy taskResolutionStrategy = null; ITaskHandler taskHandler = taskHandlerRegistry.getTaskHandler(task); String bucketId = null; ITaskProcessingPolicy taskProcessingPolicy = null; if (taskHandler == null) { log.error("No task handler found for task " + LogUtils.asParameter(task.getVersionId()) + "."); } else { taskProcessingPolicy = taskHandler.getProcessingPolicy(task); if (taskProcessingPolicy == null) { log.error("No processing policy found for task " + LogUtils.asParameter(task.getVersionId()) + "."); } else { taskResolutionStrategy = taskProcessingPolicy.getStuckTaskResolutionStrategy(task); bucketId = taskProcessingPolicy.getProcessingBucket(task); } } if (!TaskStatus.PROCESSING.name().equals(task.getStatus())) { retryTask(taskProcessingPolicy, bucketId, task, resumedCount); return; } if (taskResolutionStrategy == null) { log.error("No processing policy found for task " + LogUtils.asParameter(task.getVersionId()) + "."); markTaskAsError(bucketId, task, errorCount); return; } switch (taskResolutionStrategy) { case RETRY: retryTask(taskProcessingPolicy, bucketId, task, resumedCount); break; case MARK_AS_ERROR: markTaskAsError(bucketId, task, errorCount); break; case MARK_AS_FAILED: if (!taskDao.setStatus(task.getVersionId().getId(), TaskStatus.FAILED, task.getVersionId().getVersion())) { meterHelper.registerFailedStatusChange(task.getType(), task.getStatus(), TaskStatus.FAILED); } else { failedCount.getAndIncrement(); String taskType = task.getType(); meterHelper.registerStuckTaskMarkedAsFailed(taskType); meterHelper.registerTaskMarkedAsFailed(bucketId, taskType); } break; case IGNORE: meterHelper.registerStuckTaskAsIgnored(task.getType()); break; default: throw new UnsupportedOperationException("Resolution strategy " + taskResolutionStrategy + " is not supported."); } }  }
TasksResumer implements ITasksResumer, GracefulShutdownStrategy { protected void handleStuckTask(ITaskDao.StuckTask task, AtomicInteger resumedCount, AtomicInteger errorCount, AtomicInteger failedCount) { ITaskProcessingPolicy.StuckTaskResolutionStrategy taskResolutionStrategy = null; ITaskHandler taskHandler = taskHandlerRegistry.getTaskHandler(task); String bucketId = null; ITaskProcessingPolicy taskProcessingPolicy = null; if (taskHandler == null) { log.error("No task handler found for task " + LogUtils.asParameter(task.getVersionId()) + "."); } else { taskProcessingPolicy = taskHandler.getProcessingPolicy(task); if (taskProcessingPolicy == null) { log.error("No processing policy found for task " + LogUtils.asParameter(task.getVersionId()) + "."); } else { taskResolutionStrategy = taskProcessingPolicy.getStuckTaskResolutionStrategy(task); bucketId = taskProcessingPolicy.getProcessingBucket(task); } } if (!TaskStatus.PROCESSING.name().equals(task.getStatus())) { retryTask(taskProcessingPolicy, bucketId, task, resumedCount); return; } if (taskResolutionStrategy == null) { log.error("No processing policy found for task " + LogUtils.asParameter(task.getVersionId()) + "."); markTaskAsError(bucketId, task, errorCount); return; } switch (taskResolutionStrategy) { case RETRY: retryTask(taskProcessingPolicy, bucketId, task, resumedCount); break; case MARK_AS_ERROR: markTaskAsError(bucketId, task, errorCount); break; case MARK_AS_FAILED: if (!taskDao.setStatus(task.getVersionId().getId(), TaskStatus.FAILED, task.getVersionId().getVersion())) { meterHelper.registerFailedStatusChange(task.getType(), task.getStatus(), TaskStatus.FAILED); } else { failedCount.getAndIncrement(); String taskType = task.getType(); meterHelper.registerStuckTaskMarkedAsFailed(taskType); meterHelper.registerTaskMarkedAsFailed(bucketId, taskType); } break; case IGNORE: meterHelper.registerStuckTaskAsIgnored(task.getType()); break; default: throw new UnsupportedOperationException("Resolution strategy " + taskResolutionStrategy + " is not supported."); } }  @PostConstruct void init(); void pauseProcessing(); void resumeProcessing(); @Override void applicationStarted(); @Override void prepareForShutdown(); @Override boolean canShutdown(); }
TasksResumer implements ITasksResumer, GracefulShutdownStrategy { protected void handleStuckTask(ITaskDao.StuckTask task, AtomicInteger resumedCount, AtomicInteger errorCount, AtomicInteger failedCount) { ITaskProcessingPolicy.StuckTaskResolutionStrategy taskResolutionStrategy = null; ITaskHandler taskHandler = taskHandlerRegistry.getTaskHandler(task); String bucketId = null; ITaskProcessingPolicy taskProcessingPolicy = null; if (taskHandler == null) { log.error("No task handler found for task " + LogUtils.asParameter(task.getVersionId()) + "."); } else { taskProcessingPolicy = taskHandler.getProcessingPolicy(task); if (taskProcessingPolicy == null) { log.error("No processing policy found for task " + LogUtils.asParameter(task.getVersionId()) + "."); } else { taskResolutionStrategy = taskProcessingPolicy.getStuckTaskResolutionStrategy(task); bucketId = taskProcessingPolicy.getProcessingBucket(task); } } if (!TaskStatus.PROCESSING.name().equals(task.getStatus())) { retryTask(taskProcessingPolicy, bucketId, task, resumedCount); return; } if (taskResolutionStrategy == null) { log.error("No processing policy found for task " + LogUtils.asParameter(task.getVersionId()) + "."); markTaskAsError(bucketId, task, errorCount); return; } switch (taskResolutionStrategy) { case RETRY: retryTask(taskProcessingPolicy, bucketId, task, resumedCount); break; case MARK_AS_ERROR: markTaskAsError(bucketId, task, errorCount); break; case MARK_AS_FAILED: if (!taskDao.setStatus(task.getVersionId().getId(), TaskStatus.FAILED, task.getVersionId().getVersion())) { meterHelper.registerFailedStatusChange(task.getType(), task.getStatus(), TaskStatus.FAILED); } else { failedCount.getAndIncrement(); String taskType = task.getType(); meterHelper.registerStuckTaskMarkedAsFailed(taskType); meterHelper.registerTaskMarkedAsFailed(bucketId, taskType); } break; case IGNORE: meterHelper.registerStuckTaskAsIgnored(task.getType()); break; default: throw new UnsupportedOperationException("Resolution strategy " + taskResolutionStrategy + " is not supported."); } }  @PostConstruct void init(); void pauseProcessing(); void resumeProcessing(); @Override void applicationStarted(); @Override void prepareForShutdown(); @Override boolean canShutdown();  }