Browse Source

Improve job cancellation handling

pull/14264/head
Viacheslav Klimov 12 months ago
parent
commit
1884020a64
  1. 12
      common/data/src/main/java/org/thingsboard/server/common/data/job/JobResult.java
  2. 10
      dao/src/main/java/org/thingsboard/server/dao/job/DefaultJobService.java

12
common/data/src/main/java/org/thingsboard/server/common/data/job/JobResult.java

@ -55,11 +55,17 @@ public abstract class JobResult implements Serializable {
public void processTaskResult(TaskResult taskResult) { public void processTaskResult(TaskResult taskResult) {
if (taskResult.isSuccess()) { if (taskResult.isSuccess()) {
successfulCount++; if (totalCount == null || successfulCount < totalCount) {
successfulCount++;
}
} else if (taskResult.isDiscarded()) { } else if (taskResult.isDiscarded()) {
discardedCount++; if (totalCount == null || discardedCount < totalCount) {
discardedCount++;
}
} else { } else {
failedCount++; if (totalCount == null || failedCount < totalCount) {
failedCount++;
}
if (results.size() < 100) { // preserving only first 100 errors, not reprocessing if there are more failures if (results.size() < 100) { // preserving only first 100 errors, not reprocessing if there are more failures
results.add(taskResult); results.add(taskResult);
} }

10
dao/src/main/java/org/thingsboard/server/dao/job/DefaultJobService.java

@ -87,11 +87,7 @@ public class DefaultJobService extends AbstractEntityService implements JobServi
} }
job.getResult().setCancellationTs(System.currentTimeMillis()); job.getResult().setCancellationTs(System.currentTimeMillis());
JobStatus prevStatus = job.getStatus(); JobStatus prevStatus = job.getStatus();
if (job.getStatus() == QUEUED) { job.setStatus(CANCELLED);
job.setStatus(CANCELLED); // setting cancelled status right away, because we don't expect stats for cancelled tasks
} else if (job.getStatus() == PENDING) {
job.setStatus(RUNNING);
}
saveJob(tenantId, job, true, prevStatus); saveJob(tenantId, job, true, prevStatus);
} }
@ -145,7 +141,7 @@ public class DefaultJobService extends AbstractEntityService implements JobServi
} }
} }
if (job.getStatus() == RUNNING) { if (job.getStatus().isOneOf(RUNNING, CANCELLED)) {
if (result.getTotalCount() != null && result.getCompletedCount() >= result.getTotalCount()) { if (result.getTotalCount() != null && result.getCompletedCount() >= result.getTotalCount()) {
if (result.getCancellationTs() > 0) { if (result.getCancellationTs() > 0) {
job.setStatus(CANCELLED); job.setStatus(CANCELLED);
@ -193,7 +189,7 @@ public class DefaultJobService extends AbstractEntityService implements JobServi
private void checkWaitingJobs(TenantId tenantId, JobType jobType) { private void checkWaitingJobs(TenantId tenantId, JobType jobType) {
Job queuedJob = jobDao.findOldestByTenantIdAndTypeAndStatusForUpdate(tenantId, jobType, QUEUED); Job queuedJob = jobDao.findOldestByTenantIdAndTypeAndStatusForUpdate(tenantId, jobType, QUEUED);
if (queuedJob == null) { if (queuedJob == null || jobDao.existsByTenantIdAndTypeAndStatusOneOf(tenantId, jobType, PENDING, RUNNING)) {
return; return;
} }
queuedJob.setStatus(PENDING); queuedJob.setStatus(PENDING);

Loading…
Cancel
Save