Skip to content

Commit 76e37e7

Browse files
guowl3guowl3
committed
opt task status
1 parent 105e777 commit 76e37e7

3 files changed

Lines changed: 49 additions & 17 deletions

File tree

‎server/odc-service/src/main/java/com/oceanbase/odc/service/quartz/executor/AbstractQuartzJob.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,15 @@ public void run(JobExecutionContext context) {
105105
public OdcJob getOdcJob(JobExecutionContext context) {
106106
JobKey key = context.getJobDetail().getKey();
107107
ScheduleTaskType taskType = ScheduleTaskType.valueOf(key.getGroup());
108+
if (taskType == ScheduleTaskType.DATA_ARCHIVE && context.getResult() != null) {
109+
try {
110+
ScheduleTaskEntity taskEntity = (ScheduleTaskEntity) context.getResult();
111+
taskType = ScheduleTaskType.valueOf(taskEntity.getJobGroup());
112+
log.info("Recovery task,scheduleTaskId={}", taskEntity.getId());
113+
} catch (Exception e) {
114+
log.warn("Load history task failed,jobKey={}", key);
115+
}
116+
}
108117
switch (taskType) {
109118
case SQL_PLAN:
110119
return new SqlPlanJob();

‎server/odc-service/src/main/java/com/oceanbase/odc/service/schedule/job/DataArchiveDeleteJob.java‎

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,15 +15,18 @@
1515
*/
1616
package com.oceanbase.odc.service.schedule.job;
1717

18+
import java.util.Map;
1819
import java.util.Optional;
1920

2021
import org.quartz.JobExecutionContext;
2122

23+
import com.fasterxml.jackson.core.type.TypeReference;
2224
import com.oceanbase.odc.common.json.JsonUtils;
2325
import com.oceanbase.odc.core.shared.constant.TaskStatus;
2426
import com.oceanbase.odc.metadb.schedule.ScheduleTaskEntity;
2527
import com.oceanbase.odc.service.dlm.model.DataArchiveParameters;
2628
import com.oceanbase.odc.service.schedule.model.DataArchiveClearParameters;
29+
import com.oceanbase.odc.service.task.constants.JobParametersKeyConstants;
2730
import com.oceanbase.tools.migrator.common.enums.JobType;
2831

2932
import lombok.extern.slf4j.Slf4j;
@@ -62,11 +65,19 @@ public void executeJob(JobExecutionContext context) {
6265
scheduleTaskRepository.updateStatusById(taskEntity.getId(), TaskStatus.FAILED);
6366
return;
6467
}
65-
66-
DLMJobReq parameters = getDLMJobReq(dataArchiveTask.getJobId());
67-
parameters.setJobType(JobType.DELETE);
68-
parameters.setFireTime(context.getFireTime());
69-
parameters.setScheduleTaskId(taskEntity.getId());
68+
DLMJobReq parameters;
69+
if (taskEntity.getJobId() != null) {
70+
parameters = JsonUtils.fromJson(JsonUtils.fromJson(
71+
taskFrameworkService.find(taskEntity.getJobId()).getJobParametersJson(),
72+
new TypeReference<Map<String, String>>() {})
73+
.get(JobParametersKeyConstants.META_TASK_PARAMETER_JSON),
74+
DLMJobReq.class);
75+
} else {
76+
parameters = getDLMJobReq(dataArchiveTask.getJobId());
77+
parameters.setJobType(JobType.DELETE);
78+
parameters.setFireTime(context.getFireTime());
79+
parameters.setScheduleTaskId(taskEntity.getId());
80+
}
7081
parameters
7182
.setRateLimit(limiterService.getByOrderIdOrElseDefaultConfig(Long.parseLong(taskEntity.getJobName())));
7283
Long jobId = publishJob(parameters, dataArchiveParameters.getTimeoutMillis(),

‎server/odc-service/src/main/java/com/oceanbase/odc/service/schedule/job/DataArchiveRollbackJob.java‎

Lines changed: 24 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -15,15 +15,18 @@
1515
*/
1616
package com.oceanbase.odc.service.schedule.job;
1717

18+
import java.util.Map;
1819
import java.util.Optional;
1920

2021
import org.quartz.JobExecutionContext;
2122

23+
import com.fasterxml.jackson.core.type.TypeReference;
2224
import com.oceanbase.odc.common.json.JsonUtils;
2325
import com.oceanbase.odc.core.shared.constant.TaskStatus;
2426
import com.oceanbase.odc.metadb.schedule.ScheduleTaskEntity;
2527
import com.oceanbase.odc.service.dlm.model.DataArchiveParameters;
2628
import com.oceanbase.odc.service.schedule.model.DataArchiveRollbackParameters;
29+
import com.oceanbase.odc.service.task.constants.JobParametersKeyConstants;
2730
import com.oceanbase.tools.migrator.common.configure.DataSourceInfo;
2831
import com.oceanbase.tools.migrator.common.enums.JobType;
2932

@@ -58,20 +61,29 @@ public void executeJob(JobExecutionContext context) {
5861
DataArchiveParameters dataArchiveParameters = JsonUtils.fromJson(dataArchiveTask.getParametersJson(),
5962
DataArchiveParameters.class);
6063
// execute in task framework.
61-
DLMJobReq parameters = getDLMJobReq(dataArchiveTask.getJobId());
62-
parameters.setJobType(JobType.ROLLBACK);
63-
DataSourceInfo tempDataSource = parameters.getSourceDs();
64-
parameters.setSourceDs(parameters.getTargetDs());
65-
parameters.setTargetDs(tempDataSource);
66-
parameters.setFireTime(context.getFireTime());
64+
DLMJobReq parameters;
65+
if (taskEntity.getJobId() != null) {
66+
parameters = JsonUtils.fromJson(JsonUtils.fromJson(
67+
taskFrameworkService.find(taskEntity.getJobId()).getJobParametersJson(),
68+
new TypeReference<Map<String, String>>() {})
69+
.get(JobParametersKeyConstants.META_TASK_PARAMETER_JSON),
70+
DLMJobReq.class);
71+
} else {
72+
parameters = getDLMJobReq(dataArchiveTask.getJobId());
73+
parameters.setJobType(JobType.ROLLBACK);
74+
DataSourceInfo tempDataSource = parameters.getSourceDs();
75+
parameters.setSourceDs(parameters.getTargetDs());
76+
parameters.setTargetDs(tempDataSource);
77+
parameters.setFireTime(context.getFireTime());
78+
parameters.getTables().forEach(o -> {
79+
String temp = o.getTableName();
80+
o.setTableName(o.getTargetTableName());
81+
o.setTargetTableName(temp);
82+
});
83+
parameters.setScheduleTaskId(taskEntity.getId());
84+
}
6785
parameters
6886
.setRateLimit(limiterService.getByOrderIdOrElseDefaultConfig(Long.parseLong(taskEntity.getJobName())));
69-
parameters.getTables().forEach(o -> {
70-
String temp = o.getTableName();
71-
o.setTableName(o.getTargetTableName());
72-
o.setTargetTableName(temp);
73-
});
74-
parameters.setScheduleTaskId(taskEntity.getId());
7587
Long jobId = publishJob(parameters, dataArchiveParameters.getTimeoutMillis(),
7688
dataArchiveParameters.getSourceDatabaseId());
7789
log.info("Publish DLM job to task framework succeed,scheduleTaskId={},jobIdentity={}", taskEntity.getId(),

0 commit comments

Comments
 (0)