修复删除asinBUG
This commit is contained in:
super
@@ -1625,13 +1625,26 @@ public class DeleteBrandRunService {
|
||||
}
|
||||
resultEntity.setResultContentType(CONTENT_TYPE_XLSX);
|
||||
resultEntity.setRowCount(parsedFile.getTotalRows());
|
||||
resultEntity.setSuccess(1);
|
||||
resultEntity.setErrorMessage(null);
|
||||
fileResultMapper.updateById(resultEntity);
|
||||
boolean assembleTerminallyFailed = false;
|
||||
if (!alreadyAssembled) {
|
||||
taskFileJobService.enqueueAssembleResult(task.getId(), MODULE_TYPE, resultEntity.getId(), fileIdentity);
|
||||
waitingForAssemble = true;
|
||||
TaskFileJobEntity assembleJob = taskFileJobService.enqueueAssembleResult(task.getId(), MODULE_TYPE, resultEntity.getId(), fileIdentity);
|
||||
// job 已终态失败（retryCount 已达上限）时不再等待组装，否则任务会永远卡在 RUNNING。
|
||||
assembleTerminallyFailed = assembleJob != null
|
||||
&& "FAILED".equals(assembleJob.getStatus())
|
||||
&& assembleJob.getRetryCount() != null
|
||||
&& assembleJob.getRetryCount() >= TaskFileJobService.MAX_RETRY_COUNT;
|
||||
if (!assembleTerminallyFailed) {
|
||||
waitingForAssemble = true;
|
||||
}
|
||||
}
|
||||
if (assembleTerminallyFailed) {
|
||||
resultEntity.setSuccess(0);
|
||||
resultEntity.setErrorMessage("结果文件生成失败");
|
||||
} else {
|
||||
resultEntity.setSuccess(1);
|
||||
resultEntity.setErrorMessage(null);
|
||||
}
|
||||
fileResultMapper.updateById(resultEntity);
|
||||
|
||||
DeleteBrandResultItemVo item = new DeleteBrandResultItemVo();
|
||||
item.setResultId(resultEntity.getId());
|
||||
@@ -1878,7 +1891,11 @@ public class DeleteBrandRunService {
|
||||
Row countryRow = sheet.createRow(0);
|
||||
Row headerRow = sheet.createRow(1);
|
||||
|
||||
for (int countryIndex = 0; countryIndex < mergedFile.countries().size(); countryIndex++) {
|
||||
// SXSSF 是流式 workbook，超过窗口（200 行）的旧行会被 flush 到磁盘后不可再修改。
|
||||
// 因此这里必须按“行”顺序写入（行号严格递增），不能按“国家列”循环回头补已 flush 的行。
|
||||
int countryCount = mergedFile.countries().size();
|
||||
int maxItemCount = 0;
|
||||
for (int countryIndex = 0; countryIndex < countryCount; countryIndex++) {
|
||||
DeleteBrandProcessedCountryDto country = mergedFile.countries().get(countryIndex);
|
||||
int asinColumnIndex = countryIndex * 2;
|
||||
int statusColumnIndex = asinColumnIndex + 1;
|
||||
@@ -1888,22 +1905,26 @@ public class DeleteBrandRunService {
|
||||
createTextCell(headerRow, statusColumnIndex, "状态");
|
||||
|
||||
List<DeleteBrandCountryResultItemDto> items = country.getItems();
|
||||
if (items == null) {
|
||||
continue;
|
||||
}
|
||||
|
||||
for (int itemIndex = 0; itemIndex < items.size(); itemIndex++) {
|
||||
DeleteBrandCountryResultItemDto item = items.get(itemIndex);
|
||||
Row row = sheet.getRow(itemIndex + 2);
|
||||
if (row == null) {
|
||||
row = sheet.createRow(itemIndex + 2);
|
||||
}
|
||||
createTextCell(row, asinColumnIndex, item.getAsin());
|
||||
createTextCell(row, statusColumnIndex, item.getStatus());
|
||||
if (items != null) {
|
||||
maxItemCount = Math.max(maxItemCount, items.size());
|
||||
}
|
||||
}
|
||||
|
||||
applyDeleteBrandColumnWidths(sheet, mergedFile.countries().size() * 2);
|
||||
for (int itemIndex = 0; itemIndex < maxItemCount; itemIndex++) {
|
||||
Row row = sheet.createRow(itemIndex + 2);
|
||||
for (int countryIndex = 0; countryIndex < countryCount; countryIndex++) {
|
||||
List<DeleteBrandCountryResultItemDto> items = mergedFile.countries().get(countryIndex).getItems();
|
||||
if (items == null || itemIndex >= items.size()) {
|
||||
continue;
|
||||
}
|
||||
DeleteBrandCountryResultItemDto item = items.get(itemIndex);
|
||||
int asinColumnIndex = countryIndex * 2;
|
||||
createTextCell(row, asinColumnIndex, item.getAsin());
|
||||
createTextCell(row, asinColumnIndex + 1, item.getStatus());
|
||||
}
|
||||
}
|
||||
|
||||
applyDeleteBrandColumnWidths(sheet, countryCount * 2);
|
||||
workbook.write(outputStream);
|
||||
return outputFile;
|
||||
} catch (Exception ex) {
|
||||
|
||||
@@ -22,6 +22,7 @@ import java.util.Map;
|
||||
public class TaskFileJobService {
|
||||
|
||||
public static final String JOB_TYPE_ASSEMBLE_RESULT = "ASSEMBLE_RESULT";
|
||||
public static final int MAX_RETRY_COUNT = 5;
|
||||
|
||||
private final TaskFileJobMapper taskFileJobMapper;
|
||||
private final ApplicationEventPublisher applicationEventPublisher;
|
||||
@@ -57,6 +58,12 @@ public class TaskFileJobService {
|
||||
if ("SUCCESS".equals(existing.getStatus()) || "RUNNING".equals(existing.getStatus()) || "PENDING".equals(existing.getStatus())) {
|
||||
return existing;
|
||||
}
|
||||
// 终态失败：retryCount 已达上限的 FAILED job 不再重置为 PENDING，
|
||||
// 否则 finalize 路径会把它反复拉回排队，导致 retryCount 无限累加、任务永远卡在 RUNNING。
|
||||
int existingRetry = existing.getRetryCount() == null ? 0 : existing.getRetryCount();
|
||||
if ("FAILED".equals(existing.getStatus()) && existingRetry >= MAX_RETRY_COUNT) {
|
||||
return existing;
|
||||
}
|
||||
taskFileJobMapper.update(null, new LambdaUpdateWrapper<TaskFileJobEntity>()
|
||||
.eq(TaskFileJobEntity::getId, existing.getId())
|
||||
.set(TaskFileJobEntity::getScopeKey, scopeKey)
|
||||
@@ -72,7 +79,7 @@ public class TaskFileJobService {
|
||||
public List<TaskFileJobEntity> listRunnableJobs(int limit) {
|
||||
return taskFileJobMapper.selectList(new LambdaQueryWrapper<TaskFileJobEntity>()
|
||||
.in(TaskFileJobEntity::getStatus, List.of("PENDING", "FAILED"))
|
||||
.lt(TaskFileJobEntity::getRetryCount, 5)
|
||||
.lt(TaskFileJobEntity::getRetryCount, MAX_RETRY_COUNT)
|
||||
.orderByAsc(TaskFileJobEntity::getUpdatedAt)
|
||||
.last("limit " + Math.max(1, Math.min(limit, 100))));
|
||||
}
|
||||
@@ -87,7 +94,7 @@ public class TaskFileJobService {
|
||||
|
||||
List<TaskFileJobEntity> ownerJobs = taskFileJobMapper.selectList(new LambdaQueryWrapper<TaskFileJobEntity>()
|
||||
.in(TaskFileJobEntity::getStatus, List.of("PENDING", "FAILED"))
|
||||
.lt(TaskFileJobEntity::getRetryCount, 5)
|
||||
.lt(TaskFileJobEntity::getRetryCount, MAX_RETRY_COUNT)
|
||||
.in(TaskFileJobEntity::getModuleType, List.of("APPEARANCE_PATENT", "SIMILAR_ASIN"))
|
||||
.like(TaskFileJobEntity::getScopeKey, ownerMarker)
|
||||
.orderByAsc(TaskFileJobEntity::getUpdatedAt)
|
||||
@@ -99,7 +106,7 @@ public class TaskFileJobService {
|
||||
List<TaskFileJobEntity> jobs = new ArrayList<>(ownerJobs);
|
||||
LambdaQueryWrapper<TaskFileJobEntity> genericWrapper = new LambdaQueryWrapper<TaskFileJobEntity>()
|
||||
.in(TaskFileJobEntity::getStatus, List.of("PENDING", "FAILED"))
|
||||
.lt(TaskFileJobEntity::getRetryCount, 5)
|
||||
.lt(TaskFileJobEntity::getRetryCount, MAX_RETRY_COUNT)
|
||||
.and(wrapper -> wrapper
|
||||
.notIn(TaskFileJobEntity::getModuleType, List.of("APPEARANCE_PATENT", "SIMILAR_ASIN"))
|
||||
.or()
|
||||
@@ -160,7 +167,7 @@ public class TaskFileJobService {
|
||||
int reset = 0;
|
||||
for (TaskFileJobEntity job : jobs) {
|
||||
int retryCount = job.getRetryCount() == null ? 0 : job.getRetryCount();
|
||||
if (retryCount >= 5) {
|
||||
if (retryCount >= MAX_RETRY_COUNT) {
|
||||
taskFileJobMapper.update(null, new LambdaUpdateWrapper<TaskFileJobEntity>()
|
||||
.eq(TaskFileJobEntity::getId, job.getId())
|
||||
.eq(TaskFileJobEntity::getStatus, "RUNNING")
|
||||
@@ -267,7 +274,7 @@ public class TaskFileJobService {
|
||||
.set(TaskFileJobEntity::getRetryCount, retryCount)
|
||||
.set(TaskFileJobEntity::getErrorMessage, message == null ? "结果文件生成失败" : message)
|
||||
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now())
|
||||
.set(TaskFileJobEntity::getFinishedAt, retryCount >= 5 ? LocalDateTime.now() : null));
|
||||
.set(TaskFileJobEntity::getFinishedAt, retryCount >= MAX_RETRY_COUNT ? LocalDateTime.now() : null));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -279,7 +286,7 @@ public class TaskFileJobService {
|
||||
.eq(TaskFileJobEntity::getId, job.getId())
|
||||
.ne(TaskFileJobEntity::getStatus, "SUCCESS")
|
||||
.set(TaskFileJobEntity::getStatus, "FAILED")
|
||||
.set(TaskFileJobEntity::getRetryCount, 5)
|
||||
.set(TaskFileJobEntity::getRetryCount, MAX_RETRY_COUNT)
|
||||
.set(TaskFileJobEntity::getErrorMessage, message == null ? "结果文件生成失败" : message)
|
||||
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now())
|
||||
.set(TaskFileJobEntity::getFinishedAt, LocalDateTime.now()));
|
||||
@@ -319,11 +326,17 @@ public class TaskFileJobService {
|
||||
if (taskId == null || taskId <= 0 || moduleType == null || moduleType.isBlank()) {
|
||||
return 0L;
|
||||
}
|
||||
// 终态失败（FAILED 且 retryCount 已达上限）的 job 不能再算作“未完成”，
|
||||
// 否则主任务永远无法 finalize，会一直卡在 RUNNING。
|
||||
Long count = taskFileJobMapper.selectCount(new LambdaQueryWrapper<TaskFileJobEntity>()
|
||||
.eq(TaskFileJobEntity::getTaskId, taskId)
|
||||
.eq(TaskFileJobEntity::getModuleType, moduleType)
|
||||
.eq(TaskFileJobEntity::getJobType, JOB_TYPE_ASSEMBLE_RESULT)
|
||||
.ne(TaskFileJobEntity::getStatus, "SUCCESS"));
|
||||
.ne(TaskFileJobEntity::getStatus, "SUCCESS")
|
||||
.and(wrapper -> wrapper
|
||||
.ne(TaskFileJobEntity::getStatus, "FAILED")
|
||||
.or()
|
||||
.lt(TaskFileJobEntity::getRetryCount, MAX_RETRY_COUNT)));
|
||||
return count == null ? 0L : count;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user