fix(delete-brand): 分片缺失降级组装收尾+失败兜底生成可下载结果
- 结果分片缺失 ≤ missingChunkFallbackLimit(默认20) 时降级组装,缺失行标「未回传」,任务不再整体白跑
- 超阈值失败/组装异常时通过 ResultFileJobHandler.fallbackAssembleOnFailure 兜底生成部分结果文件
- 组装失败文案带缺失分片号明细(如 缺失分片: [247](已收 4283/4284))
- 新增 GET /api/delete-brand/tasks/{taskId}/missing-chunks,补传后自动恢复终态失败组装 job
- stale-check 失败文案改为含最后心跳时间与最近组装失败真实原因(修复误导性「回传长时间无响应」)
- 新增降级合并/缺失分片单测,Handler/Worker 日志审计 stage 白名单补 FALLBACK
This commit is contained in:
@@ -44,4 +44,11 @@ public class DeleteBrandProgressProperties {
|
||||
* Withdraw RUNNING timeout auto-finalize/fail threshold in minutes.
|
||||
*/
|
||||
private long withdrawStaleTimeoutMinutes = 30;
|
||||
|
||||
/**
|
||||
* 组装结果文件时允许缺失的最大分片数:缺失不超过该值时降级组装(缺失行状态写"未回传"),
|
||||
* 保证任务不因少量分片丢失整体白跑;超出时任务失败,但仍会在重试耗尽后尝试
|
||||
* fallbackAssembleOnFailure 生成部分结果文件。
|
||||
*/
|
||||
private int missingChunkFallbackLimit = 20;
|
||||
}
|
||||
|
||||
+7
@@ -129,6 +129,13 @@ public class DeleteBrandRunController {
|
||||
}
|
||||
}
|
||||
|
||||
@GetMapping("/tasks/{taskId}/missing-chunks")
|
||||
@Operation(summary = "查询删除品牌任务缺失的分片号", description = "供插件端在组装失败后补传缺失分片:携带缺失 chunkIndex 重新 POST 同一 /tasks/{taskId}/result 即可触发恢复组装。")
|
||||
public ApiResponse<com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandMissingChunksVo> missingChunks(
|
||||
@PathVariable Long taskId) {
|
||||
return ApiResponse.success(deleteBrandRunService.listMissingChunks(taskId));
|
||||
}
|
||||
|
||||
@GetMapping("/tasks/{taskId}/download")
|
||||
@Operation(summary = "下载删除品牌任务结果", description = "返回任务结果文件流,便于桌面端通过 save_file_from_url_new 直接保存。")
|
||||
public void download(@PathVariable Long taskId, jakarta.servlet.http.HttpServletResponse response) {
|
||||
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
package com.nanri.aiimage.modules.deletebrand.model.vo;
|
||||
|
||||
import io.swagger.v3.oas.annotations.media.Schema;
|
||||
import lombok.Data;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 删除品牌任务缺失分片查询结果(供插件端补传)。
|
||||
*/
|
||||
@Data
|
||||
@Schema(description = "删除品牌任务缺失分片")
|
||||
public class DeleteBrandMissingChunksVo {
|
||||
|
||||
@Schema(description = "任务ID")
|
||||
private Long taskId;
|
||||
|
||||
@Schema(description = "按文件维度的缺失分片列表")
|
||||
private List<MissingFileItem> files = new ArrayList<>();
|
||||
|
||||
@Data
|
||||
@Schema(description = "单个文件的缺失分片")
|
||||
public static class MissingFileItem {
|
||||
@Schema(description = "文件标识(hash)")
|
||||
private String fileIdentity;
|
||||
|
||||
@Schema(description = "文件总分片数")
|
||||
private Integer chunkTotal;
|
||||
|
||||
@Schema(description = "已回传分片数")
|
||||
private Integer received;
|
||||
|
||||
@Schema(description = "缺失的分片号(1 起)")
|
||||
private List<Integer> missing = new ArrayList<>();
|
||||
}
|
||||
}
|
||||
+6
@@ -30,4 +30,10 @@ public class DeleteBrandResultFileJobHandler implements ResultFileJobHandler {
|
||||
public void cleanup(TaskFileJobEntity job) {
|
||||
deleteBrandRunService.cleanupResultFileJob(job);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean fallbackAssembleOnFailure(TaskFileJobEntity job, String message) {
|
||||
// 分片缺失等导致组装 job 重试耗尽时:忽略缺失阈值,尽力用已回传数据生成部分结果文件供下载。
|
||||
return deleteBrandRunService.fallbackAssembleResultFile(job);
|
||||
}
|
||||
}
|
||||
|
||||
+184
-12
@@ -21,6 +21,7 @@ import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandCountryGroupVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandHistoryVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandLineProgressInfoVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandLineProgressVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandMissingChunksVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandPreviewRowVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandResultItemVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandRunVo;
|
||||
@@ -78,6 +79,9 @@ public class DeleteBrandRunService {
|
||||
|
||||
private static final String MODULE_TYPE = "DELETE_BRAND";
|
||||
|
||||
/** 降级组装时缺失分片对应的结果状态(客户端未回传该行)。 */
|
||||
static final String MISSING_CHUNK_STATUS = "未回传";
|
||||
|
||||
public TaskProgressLightBatchVo progressLight(List<Long> taskIds) {
|
||||
return taskProgressLightAssembler.assemble(MODULE_TYPE, null, taskIds);
|
||||
}
|
||||
@@ -1342,6 +1346,8 @@ public class DeleteBrandRunService {
|
||||
progress.put("last_heartbeat_at", String.valueOf(System.currentTimeMillis()));
|
||||
deleteBrandTaskCacheService.saveProgress(taskId, progress, true);
|
||||
|
||||
maybeRecoverAssembleOnCompletedChunks(taskId, fileIdentity, parsedFile);
|
||||
|
||||
if (isFileCompleted(fileDto)) {
|
||||
log.info("[DeleteBrand] File chunk transmission fully reached via chunkTotal for taskId: {}, file: {}", taskId, fileDto.getSourceFilename());
|
||||
final String identityForTx = fileIdentity;
|
||||
@@ -1357,6 +1363,39 @@ public class DeleteBrandRunService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 补传恢复:某文件的分片此前因缺失导致组装 job 终态失败(重试耗尽),
|
||||
* 客户端补传缺口后这里把失败的组装 job 重置为 PENDING 重新派发。
|
||||
* 常态(无终态失败 job)下只查一次 job 状态即返回,不触发分片扫描。
|
||||
*/
|
||||
private void maybeRecoverAssembleOnCompletedChunks(Long taskId, String fileIdentity, DeleteBrandParsedFileCacheDto parsedFile) {
|
||||
if (taskId == null || fileIdentity == null || fileIdentity.isBlank()) {
|
||||
return;
|
||||
}
|
||||
FileResultEntity resultEntity = findResultEntity(taskId, fileIdentity, parsedFile);
|
||||
if (resultEntity == null) {
|
||||
return;
|
||||
}
|
||||
if (taskFileJobService.hasSuccessfulAssembleJob(taskId, MODULE_TYPE, resultEntity.getId())) {
|
||||
return;
|
||||
}
|
||||
if (!taskFileJobService.isTerminalFailedAssembleJob(taskId, MODULE_TYPE, resultEntity.getId())) {
|
||||
return;
|
||||
}
|
||||
List<DeleteBrandResultFileDto> chunks = deleteBrandTaskStorageService.loadMergedChunks(taskId).getOrDefault(fileIdentity, List.of());
|
||||
if (!isMergedFileCompleted(chunks)) {
|
||||
return;
|
||||
}
|
||||
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||
if (task == null) {
|
||||
return;
|
||||
}
|
||||
log.info("[DeleteBrand] missing chunks patched, recovering terminal failed assemble -> taskId={} file={} resultId={}",
|
||||
taskId, fileIdentity, resultEntity.getId());
|
||||
taskFileJobService.resetTerminalFailedForRecovery(taskId, MODULE_TYPE, resultEntity.getId());
|
||||
enqueueCompletedFileAssembleJob(task, fileIdentity, parsedFile);
|
||||
}
|
||||
|
||||
private void enqueueCompletedFileAssembleJob(FileTaskEntity task,
|
||||
String fileIdentity,
|
||||
DeleteBrandParsedFileCacheDto parsedFile) {
|
||||
@@ -1558,6 +1597,30 @@ public class DeleteBrandRunService {
|
||||
return true;
|
||||
}
|
||||
|
||||
/** 缺失的分片号(1..chunkTotal 中未回传的序号),升序返回;chunkTotal 未知时返回空表。 */
|
||||
private List<Integer> missingChunkIndexes(List<DeleteBrandResultFileDto> chunks) {
|
||||
if (chunks == null || chunks.isEmpty()) {
|
||||
return List.of();
|
||||
}
|
||||
Integer chunkTotal = chunks.get(0).getChunkTotal();
|
||||
if (chunkTotal == null || chunkTotal <= 0) {
|
||||
return List.of();
|
||||
}
|
||||
Set<Integer> indexes = new LinkedHashSet<>();
|
||||
for (DeleteBrandResultFileDto chunk : chunks) {
|
||||
if (chunk.getChunkIndex() != null && chunkTotal.equals(chunk.getChunkTotal())) {
|
||||
indexes.add(chunk.getChunkIndex());
|
||||
}
|
||||
}
|
||||
List<Integer> missing = new ArrayList<>();
|
||||
for (int i = 1; i <= chunkTotal; i++) {
|
||||
if (!indexes.contains(i)) {
|
||||
missing.add(i);
|
||||
}
|
||||
}
|
||||
return missing;
|
||||
}
|
||||
|
||||
private void finalizeTask(FileTaskEntity task,
|
||||
Map<String, DeleteBrandParsedFileCacheDto> parsedPayload,
|
||||
Map<String, List<DeleteBrandResultFileDto>> mergedByFile) {
|
||||
@@ -1591,10 +1654,19 @@ public class DeleteBrandRunService {
|
||||
String fileIdentity = entry.getKey();
|
||||
DeleteBrandParsedFileCacheDto parsedFile = entry.getValue();
|
||||
List<DeleteBrandResultFileDto> chunks = mergedByFile.get(fileIdentity);
|
||||
if (!isMergedFileCompleted(chunks)) {
|
||||
throw new BusinessException("存在未完成的结果分片: " + parsedFile.getSourceFilename());
|
||||
boolean chunksComplete = isMergedFileCompleted(chunks);
|
||||
if (!chunksComplete) {
|
||||
List<Integer> missing = missingChunkIndexes(chunks);
|
||||
int limit = Math.max(1, deleteBrandProgressProperties.getMissingChunkFallbackLimit());
|
||||
if (missing.size() > limit) {
|
||||
throw new BusinessException("存在未完成的结果分片: " + parsedFile.getSourceFilename() + ",缺失分片: " + missing
|
||||
+ "(已收 " + (chunks == null ? 0 : chunks.size()) + "/"
|
||||
+ (chunks == null || chunks.isEmpty() || chunks.get(0).getChunkTotal() == null ? 0 : chunks.get(0).getChunkTotal()) + ")");
|
||||
}
|
||||
log.warn("[DeleteBrand] finalize assemble with missing chunks -> taskId={} file={} missing={} fallbackLimit={}",
|
||||
task.getId(), parsedFile.getSourceFilename(), missing, limit);
|
||||
}
|
||||
mergeChunks(parsedFile, chunks);
|
||||
mergeChunks(parsedFile, chunks, !chunksComplete);
|
||||
|
||||
FileResultEntity resultEntity = resultEntityByIdentity.get(fileIdentity);
|
||||
if (resultEntity == null) {
|
||||
@@ -1697,6 +1769,15 @@ public class DeleteBrandRunService {
|
||||
}
|
||||
|
||||
public void processResultFileJob(TaskFileJobEntity job) {
|
||||
processResultFileJob(job, false);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理结果文件组装 job。
|
||||
*
|
||||
* @param forceFallback 失败兜底模式:忽略缺失阈值,缺失分片一律降级为"未回传"并生成部分结果文件
|
||||
*/
|
||||
public void processResultFileJob(TaskFileJobEntity job, boolean forceFallback) {
|
||||
if (job == null || job.getTaskId() == null || job.getResultId() == null) {
|
||||
throw new BusinessException("结果文件任务参数不完整");
|
||||
}
|
||||
@@ -1722,10 +1803,20 @@ public class DeleteBrandRunService {
|
||||
throw new BusinessException("任务原始数据不存在: " + fileIdentity);
|
||||
}
|
||||
List<DeleteBrandResultFileDto> chunks = mergedByFile.get(fileIdentity);
|
||||
if (!isMergedFileCompleted(chunks)) {
|
||||
throw new BusinessException("结果分片未完整: " + fileIdentity);
|
||||
boolean chunksComplete = isMergedFileCompleted(chunks);
|
||||
if (!chunksComplete) {
|
||||
List<Integer> missing = missingChunkIndexes(chunks);
|
||||
int limit = Math.max(1, deleteBrandProgressProperties.getMissingChunkFallbackLimit());
|
||||
if (!forceFallback && missing.size() > limit) {
|
||||
int chunkTotal = chunks == null || chunks.isEmpty() || chunks.get(0).getChunkTotal() == null
|
||||
? 0 : chunks.get(0).getChunkTotal();
|
||||
throw new BusinessException("结果分片未完整: " + fileIdentity + ",缺失分片: " + missing
|
||||
+ "(已收 " + (chunks == null ? 0 : chunks.size()) + "/" + chunkTotal + ")");
|
||||
}
|
||||
log.warn("[DeleteBrand] result chunks incomplete, fallback assemble -> taskId={} file={} missing={} received={} forceFallback={} fallbackLimit={}",
|
||||
job.getTaskId(), fileIdentity, missing, chunks == null ? 0 : chunks.size(), forceFallback, limit);
|
||||
}
|
||||
MergedDeleteBrandFile mergedFile = mergeChunks(parsedFile, chunks);
|
||||
MergedDeleteBrandFile mergedFile = mergeChunks(parsedFile, chunks, !chunksComplete);
|
||||
File outputFile = buildResultWorkbookPreserveLayout(job.getTaskId(), parsedFile, mergedFile);
|
||||
try {
|
||||
deleteBrandTaskCacheService.saveProgress(job.getTaskId(), Map.of(
|
||||
@@ -1758,6 +1849,55 @@ public class DeleteBrandRunService {
|
||||
cleanupCompletedTaskDataIfNoPendingFileJobs(job.getTaskId());
|
||||
}
|
||||
|
||||
/**
|
||||
* 失败兜底:job 重试耗尽后调用,忽略缺失阈值、尽力用已回传分片生成部分结果文件。
|
||||
* 失败只记日志(job 已终态失败,不改变其状态),返回是否有可下载结果。
|
||||
*/
|
||||
public boolean fallbackAssembleResultFile(TaskFileJobEntity job) {
|
||||
if (job == null || job.getTaskId() == null) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
processResultFileJob(job, true);
|
||||
log.warn("[DeleteBrand] fallback assemble produced partial result -> taskId={} resultId={} file={}",
|
||||
job.getTaskId(), job.getResultId(), job.getScopeKey());
|
||||
return true;
|
||||
} catch (Exception ex) {
|
||||
log.warn("[DeleteBrand] fallback assemble also failed taskId={} resultId={} msg={}",
|
||||
job.getTaskId(), job.getResultId(), ex.getMessage());
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 查询任务每个文件缺失的分片号,供插件端补传(POST 同一 /tasks/{taskId}/result 携带对应 chunkIndex)。
|
||||
*/
|
||||
public DeleteBrandMissingChunksVo listMissingChunks(Long taskId) {
|
||||
DeleteBrandMissingChunksVo vo = new DeleteBrandMissingChunksVo();
|
||||
vo.setTaskId(taskId);
|
||||
if (taskId == null || taskId <= 0) {
|
||||
return vo;
|
||||
}
|
||||
Map<String, List<DeleteBrandResultFileDto>> mergedByFile = deleteBrandTaskStorageService.loadMergedChunks(taskId);
|
||||
List<DeleteBrandMissingChunksVo.MissingFileItem> files = new ArrayList<>();
|
||||
for (Map.Entry<String, List<DeleteBrandResultFileDto>> entry : mergedByFile.entrySet()) {
|
||||
List<Integer> missing = missingChunkIndexes(entry.getValue());
|
||||
if (missing.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
DeleteBrandMissingChunksVo.MissingFileItem item = new DeleteBrandMissingChunksVo.MissingFileItem();
|
||||
item.setFileIdentity(entry.getKey());
|
||||
item.setChunkTotal(entry.getValue().isEmpty() || entry.getValue().get(0).getChunkTotal() == null
|
||||
? 0 : entry.getValue().get(0).getChunkTotal());
|
||||
item.setReceived(entry.getValue().size());
|
||||
item.setMissing(missing);
|
||||
files.add(item);
|
||||
}
|
||||
vo.setFiles(files);
|
||||
log.info("[DeleteBrand] missing chunks queried taskId={} files={}", taskId, files.size());
|
||||
return vo;
|
||||
}
|
||||
|
||||
private void cleanupCompletedTaskDataIfNoPendingFileJobs(Long taskId) {
|
||||
if (taskId == null || taskId <= 0) {
|
||||
return;
|
||||
@@ -1780,7 +1920,31 @@ public class DeleteBrandRunService {
|
||||
return filename;
|
||||
}
|
||||
|
||||
private MergedDeleteBrandFile mergeChunks(DeleteBrandParsedFileCacheDto parsedFile, List<DeleteBrandResultFileDto> chunks) {
|
||||
/** 没有任何已回传分片时的降级合并:所有国家/ASIN 行标记"未回传",布局与全量结果一致。 */
|
||||
private MergedDeleteBrandFile mergeChunksAsMissingAll(DeleteBrandParsedFileCacheDto parsedFile) {
|
||||
List<DeleteBrandProcessedCountryDto> mergedCountries = new ArrayList<>();
|
||||
for (DeleteBrandCountryGroupVo parsedCountry : parsedFile.getCountries()) {
|
||||
DeleteBrandProcessedCountryDto mergedCountry = new DeleteBrandProcessedCountryDto();
|
||||
mergedCountry.setCountry(parsedCountry.getCountry());
|
||||
for (DeleteBrandCountryAsinVo parsedAsin : parsedCountry.getItems()) {
|
||||
DeleteBrandCountryResultItemDto row = new DeleteBrandCountryResultItemDto();
|
||||
row.setAsin(parsedAsin.getAsin());
|
||||
row.setStatus(MISSING_CHUNK_STATUS);
|
||||
mergedCountry.getItems().add(row);
|
||||
}
|
||||
mergedCountries.add(mergedCountry);
|
||||
}
|
||||
return new MergedDeleteBrandFile("", "", 0, mergedCountries);
|
||||
}
|
||||
|
||||
private MergedDeleteBrandFile mergeChunks(DeleteBrandParsedFileCacheDto parsedFile, List<DeleteBrandResultFileDto> chunks, boolean allowMissing) {
|
||||
if (chunks == null || chunks.isEmpty()) {
|
||||
// 没有任何已回传分片:降级时全部行标记"未回传",严格模式抛业务异常(而非索引越界)
|
||||
if (!allowMissing) {
|
||||
throw new BusinessException("缺少国家结果: " + parsedFile.getSourceFilename());
|
||||
}
|
||||
return mergeChunksAsMissingAll(parsedFile);
|
||||
}
|
||||
Integer chunkTotal = chunks.get(0).getChunkTotal();
|
||||
for (DeleteBrandResultFileDto chunk : chunks) {
|
||||
if (!chunkTotal.equals(chunk.getChunkTotal())) {
|
||||
@@ -1804,18 +1968,26 @@ public class DeleteBrandRunService {
|
||||
for (DeleteBrandCountryGroupVo parsedCountry : parsedFile.getCountries()) {
|
||||
Map<String, DeleteBrandCountryResultItemDto> processedItems = processedByCountry.get(parsedCountry.getCountry());
|
||||
if (processedItems == null) {
|
||||
throw new BusinessException("缺少国家结果: " + parsedFile.getSourceFilename() + " / " + parsedCountry.getCountry());
|
||||
// 允许缺失时整国家降级:全部行标记"未回传",否则文件缺口会让用户误以为已删除。
|
||||
if (!allowMissing) {
|
||||
throw new BusinessException("缺少国家结果: " + parsedFile.getSourceFilename() + " / " + parsedCountry.getCountry());
|
||||
}
|
||||
processedItems = Map.of();
|
||||
}
|
||||
DeleteBrandProcessedCountryDto mergedCountry = new DeleteBrandProcessedCountryDto();
|
||||
mergedCountry.setCountry(parsedCountry.getCountry());
|
||||
for (DeleteBrandCountryAsinVo parsedAsin : parsedCountry.getItems()) {
|
||||
DeleteBrandCountryResultItemDto processed = processedItems.get(normalizeAsinKey(parsedAsin.getAsin()));
|
||||
if (processed == null) {
|
||||
throw new BusinessException("缺少 ASIN 结果: " + parsedFile.getSourceFilename() + " / " + parsedCountry.getCountry() + " / " + parsedAsin.getAsin());
|
||||
}
|
||||
DeleteBrandCountryResultItemDto row = new DeleteBrandCountryResultItemDto();
|
||||
row.setAsin(parsedAsin.getAsin());
|
||||
row.setStatus(processed.getStatus());
|
||||
if (processed == null) {
|
||||
if (!allowMissing) {
|
||||
throw new BusinessException("缺少 ASIN 结果: " + parsedFile.getSourceFilename() + " / " + parsedCountry.getCountry() + " / " + parsedAsin.getAsin());
|
||||
}
|
||||
row.setStatus(MISSING_CHUNK_STATUS);
|
||||
} else {
|
||||
row.setStatus(processed.getStatus());
|
||||
}
|
||||
mergedCountry.getItems().add(row);
|
||||
}
|
||||
mergedCountries.add(mergedCountry);
|
||||
|
||||
+22
-1
@@ -257,7 +257,7 @@ public class DeleteBrandStaleTaskService {
|
||||
.eq(FileTaskEntity::getModuleType, MODULE_TYPE_DELETE_BRAND)
|
||||
.eq(FileTaskEntity::getStatus, "RUNNING")
|
||||
.set(FileTaskEntity::getStatus, "FAILED")
|
||||
.set(FileTaskEntity::getErrorMessage, "结果回传长时间无响应,任务已自动失败")
|
||||
.set(FileTaskEntity::getErrorMessage, buildStaleFailReason(task, entry, minutes, nowMillis))
|
||||
.set(FileTaskEntity::getUpdatedAt, LocalDateTime.now())
|
||||
.set(FileTaskEntity::getFinishedAt, LocalDateTime.now()));
|
||||
|
||||
@@ -272,6 +272,27 @@ public class DeleteBrandStaleTaskService {
|
||||
boolean hasStartedProgress, boolean finalizeThrew) {
|
||||
}
|
||||
|
||||
/**
|
||||
* 组装超时失败文案:带最后心跳时间与最近一次组装失败的真实原因,
|
||||
* 避免"分片缺失导致组装失败"被误报成"客户端长时间无响应"。
|
||||
*/
|
||||
private String buildStaleFailReason(FileTaskEntity task, StaleFinalizeEntry entry, long minutes, long nowMillis) {
|
||||
StringBuilder sb = new StringBuilder("结果回传中断超过 ").append(minutes).append(" 分钟");
|
||||
if (entry.lastHeartbeatAt > 0L) {
|
||||
long elapsedMinutes = Math.max(1L, (nowMillis - entry.lastHeartbeatAt) / 60000L);
|
||||
sb.append("(最后心跳 ").append(Instant.ofEpochMilli(entry.lastHeartbeatAt))
|
||||
.append(",距今 ").append(elapsedMinutes).append(" 分钟)");
|
||||
} else {
|
||||
sb.append("(从未收到心跳或结果分片)");
|
||||
}
|
||||
String assembleError = taskFileJobService.latestFailedAssembleError(task.getId(), MODULE_TYPE_DELETE_BRAND);
|
||||
if (assembleError != null && !assembleError.isBlank()) {
|
||||
sb.append(";最近一次组装失败:").append(assembleError);
|
||||
}
|
||||
sb.append(",任务已自动失败(可下载的部分结果见任务结果文件,或重新提交任务)");
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
private ProductRiskStaleCheckStats failStaleProductRiskResolveTasks() {
|
||||
ProductRiskStaleCheckStats stats = new ProductRiskStaleCheckStats();
|
||||
long minutes = Math.max(1L, deleteBrandProgressProperties.getProductRiskStaleTimeoutMinutes());
|
||||
|
||||
+10
@@ -27,6 +27,16 @@ public interface ResultFileJobHandler {
|
||||
default void onFailure(TaskFileJobEntity job, String message) {
|
||||
}
|
||||
|
||||
/**
|
||||
* 失败兜底:job 重试耗尽(终态 FAILED)时,用已回传的部分数据尝试生成部分结果文件,
|
||||
* 让用户仍能下载已做好的部分(缺失行由业务侧降级填充"未回传"等占位状态)。
|
||||
* 实现应自吞异常(兜底失败只记日志,不影响 job 已终态失败的事实);
|
||||
* 返回 true 表示已生成可下载的结果文件。
|
||||
*/
|
||||
default boolean fallbackAssembleOnFailure(TaskFileJobEntity job, String message) {
|
||||
return false;
|
||||
}
|
||||
|
||||
default boolean supportsAsyncOffload() {
|
||||
return false;
|
||||
}
|
||||
|
||||
+62
@@ -6,6 +6,7 @@ import com.nanri.aiimage.modules.task.mapper.TaskFileJobMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskFileJobDispatchEvent;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.dao.DuplicateKeyException;
|
||||
import org.springframework.stereotype.Service;
|
||||
@@ -21,6 +22,7 @@ import java.util.stream.Collectors;
|
||||
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
@Slf4j
|
||||
public class TaskFileJobService {
|
||||
|
||||
public static final String JOB_TYPE_ASSEMBLE_RESULT = "ASSEMBLE_RESULT";
|
||||
@@ -88,6 +90,66 @@ public class TaskFileJobService {
|
||||
return refreshed;
|
||||
}
|
||||
|
||||
/**
|
||||
* 该 result 是否已有重试耗尽的终态失败组装 job(分片缺失类失败后用于补传恢复判断)。
|
||||
*/
|
||||
public boolean isTerminalFailedAssembleJob(Long taskId, String moduleType, Long resultId) {
|
||||
TaskFileJobEntity existing = findJob(taskId, moduleType, resultId, JOB_TYPE_ASSEMBLE_RESULT);
|
||||
if (existing == null || !"FAILED".equals(existing.getStatus())) {
|
||||
return false;
|
||||
}
|
||||
int retry = existing.getRetryCount() == null ? 0 : existing.getRetryCount();
|
||||
return retry >= MAX_RETRY_COUNT;
|
||||
}
|
||||
|
||||
/**
|
||||
* 补传完整后恢复终态失败的组装 job:重置为 PENDING(retryCount 清零)重新派发。
|
||||
* 幂等:job 不存在、非 FAILED 或未达重试上限时不动作,返回当前 job。
|
||||
*/
|
||||
public TaskFileJobEntity resetTerminalFailedForRecovery(Long taskId, String moduleType, Long resultId) {
|
||||
TaskFileJobEntity existing = findJob(taskId, moduleType, resultId, JOB_TYPE_ASSEMBLE_RESULT);
|
||||
if (existing == null || !"FAILED".equals(existing.getStatus())) {
|
||||
return existing;
|
||||
}
|
||||
int retry = existing.getRetryCount() == null ? 0 : existing.getRetryCount();
|
||||
if (retry < MAX_RETRY_COUNT) {
|
||||
return existing;
|
||||
}
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
taskFileJobMapper.update(null, new LambdaUpdateWrapper<TaskFileJobEntity>()
|
||||
.eq(TaskFileJobEntity::getId, existing.getId())
|
||||
.set(TaskFileJobEntity::getStatus, "PENDING")
|
||||
.set(TaskFileJobEntity::getRetryCount, 0)
|
||||
.set(TaskFileJobEntity::getErrorMessage, null)
|
||||
.set(TaskFileJobEntity::getFinishedAt, null)
|
||||
.set(TaskFileJobEntity::getTerminalCallbackAt, null)
|
||||
.set(TaskFileJobEntity::getUpdatedAt, now));
|
||||
log.info("[task-file-job] terminal failed assemble job reset for recovery jobId={} taskId={} moduleType={} resultId={}",
|
||||
existing.getId(), taskId, moduleType, resultId);
|
||||
TaskFileJobEntity refreshed = taskFileJobMapper.selectById(existing.getId());
|
||||
publishDispatchEvent(refreshed);
|
||||
return refreshed;
|
||||
}
|
||||
|
||||
/**
|
||||
* 最近一次失败组装 job 的错误信息(用于任务超时失败文案里拼接真实原因)。
|
||||
*/
|
||||
public String latestFailedAssembleError(Long taskId, String moduleType) {
|
||||
if (taskId == null || moduleType == null || moduleType.isBlank()) {
|
||||
return null;
|
||||
}
|
||||
TaskFileJobEntity job = taskFileJobMapper.selectOne(new LambdaQueryWrapper<TaskFileJobEntity>()
|
||||
.eq(TaskFileJobEntity::getTaskId, taskId)
|
||||
.eq(TaskFileJobEntity::getModuleType, moduleType)
|
||||
.eq(TaskFileJobEntity::getJobType, JOB_TYPE_ASSEMBLE_RESULT)
|
||||
.eq(TaskFileJobEntity::getStatus, "FAILED")
|
||||
.orderByDesc(TaskFileJobEntity::getId)
|
||||
.last("limit 1"));
|
||||
return job == null || job.getErrorMessage() == null || job.getErrorMessage().isBlank()
|
||||
? null
|
||||
: job.getErrorMessage();
|
||||
}
|
||||
|
||||
public List<TaskFileJobEntity> listRunnableJobs(int limit) {
|
||||
return taskFileJobMapper.selectList(new LambdaQueryWrapper<TaskFileJobEntity>()
|
||||
.in(TaskFileJobEntity::getStatus, List.of("PENDING", "FAILED"))
|
||||
|
||||
+10
@@ -276,6 +276,16 @@ public class TaskResultFileJobWorker {
|
||||
ResultFileJobHandler handler = handlerRegistry.asMap().get(job.getModuleType());
|
||||
if (handler != null) {
|
||||
handler.onFailure(job, message);
|
||||
try {
|
||||
if (handler.fallbackAssembleOnFailure(job, message)) {
|
||||
log.info("[task-file-job] retry-exhausted fallback assemble produced downloadable result jobId={} taskId={} moduleType={} resultId={} stage=FALLBACK",
|
||||
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId());
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
// 兜底失败不影响 job 终态;下次 stuck/exhausted 扫描会重入(幂等),持续失败仅记日志。
|
||||
log.warn("[task-file-job] fallback assemble threw jobId={} taskId={} moduleType={} resultId={} msg={} stage=FALLBACK",
|
||||
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), ex.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+1
-1
@@ -100,7 +100,7 @@ class TaskResultFileJobWorkerLogAuditTest {
|
||||
}
|
||||
// 阶段字面量来自允许集合(防新增阶段拼写漂移)
|
||||
List<String> allowed = List.of("SKIP_OWNER", "REQUEUE", "WAIT_LLM", "DEFER", "SUCCESS",
|
||||
"ORPHAN", "FAILED", "FINALIZE", "HEARTBEAT");
|
||||
"ORPHAN", "FAILED", "FINALIZE", "HEARTBEAT", "FALLBACK");
|
||||
for (String l : perJob) {
|
||||
if (l.contains("stage=")) {
|
||||
int idx = l.indexOf("stage=");
|
||||
|
||||
+188
@@ -0,0 +1,188 @@
|
||||
package com.nanri.aiimage.modules.deletebrand.service;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.nanri.aiimage.common.exception.BusinessException;
|
||||
import com.nanri.aiimage.config.DeleteBrandProgressProperties;
|
||||
import com.nanri.aiimage.config.TaskPressureProperties;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.cache.DeleteBrandParsedFileCacheDto;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandCountryResultItemDto;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandProcessedCountryDto;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandResultFileDto;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandCountryAsinVo;
|
||||
import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandCountryGroupVo;
|
||||
import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
|
||||
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
|
||||
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
||||
import com.nanri.aiimage.modules.task.service.TaskProgressLightAssembler;
|
||||
import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
/**
|
||||
* 删除品牌分片缺失降级组装(修复:缺分片不整任务白跑)纯逻辑单测。
|
||||
*/
|
||||
class DeleteBrandRunServiceFallbackTest {
|
||||
|
||||
private final DeleteBrandRunService service = new DeleteBrandRunService(
|
||||
mock(FileTaskMapper.class),
|
||||
mock(FileResultMapper.class),
|
||||
mock(LocalFileStorageService.class),
|
||||
mock(DeleteBrandTaskCacheService.class),
|
||||
mock(DeleteBrandTaskStorageService.class),
|
||||
mock(ZiniaoShopSwitchService.class),
|
||||
mock(OssStorageService.class),
|
||||
mock(ObjectMapper.class),
|
||||
mock(DeleteBrandProgressProperties.class),
|
||||
mock(TaskPressureProperties.class),
|
||||
mock(TaskDistributedLockService.class),
|
||||
mock(TaskFileJobService.class),
|
||||
mock(PlatformTransactionManager.class),
|
||||
mock(TaskProgressLightAssembler.class));
|
||||
|
||||
private DeleteBrandResultFileDto chunk(int index, int total, String asin, String status) {
|
||||
DeleteBrandResultFileDto dto = new DeleteBrandResultFileDto();
|
||||
dto.setFileKey("scope-key");
|
||||
dto.setSourceFilename("魏振峰.xlsx");
|
||||
dto.setChunkIndex(index);
|
||||
dto.setChunkTotal(total);
|
||||
dto.setProcessedRows(index);
|
||||
dto.setTotalRows(total);
|
||||
dto.setCurrentCountry("US");
|
||||
dto.setCurrentAsin(asin);
|
||||
DeleteBrandProcessedCountryDto country = new DeleteBrandProcessedCountryDto();
|
||||
country.setCountry("US");
|
||||
DeleteBrandCountryResultItemDto item = new DeleteBrandCountryResultItemDto();
|
||||
item.setAsin(asin);
|
||||
item.setStatus(status);
|
||||
country.setItems(List.of(item));
|
||||
dto.setCountries(List.of(country));
|
||||
return dto;
|
||||
}
|
||||
|
||||
private DeleteBrandParsedFileCacheDto parsedFile(String... asins) {
|
||||
DeleteBrandParsedFileCacheDto parsed = new DeleteBrandParsedFileCacheDto();
|
||||
parsed.setFileKey("scope-key");
|
||||
parsed.setSourceFilename("魏振峰.xlsx");
|
||||
parsed.setTotalRows(asins.length);
|
||||
DeleteBrandCountryGroupVo country = new DeleteBrandCountryGroupVo();
|
||||
country.setCountry("US");
|
||||
for (String asin : asins) {
|
||||
DeleteBrandCountryAsinVo asinVo = new DeleteBrandCountryAsinVo();
|
||||
asinVo.setAsin(asin);
|
||||
country.getItems().add(asinVo);
|
||||
}
|
||||
parsed.setCountries(List.of(country));
|
||||
return parsed;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private List<Integer> missingChunkIndexes(List<DeleteBrandResultFileDto> chunks) {
|
||||
return (List<Integer>) ReflectionTestUtils.invokeMethod(service, "missingChunkIndexes", chunks);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private Object mergeChunks(DeleteBrandParsedFileCacheDto parsedFile, List<DeleteBrandResultFileDto> chunks, boolean allowMissing) {
|
||||
return ReflectionTestUtils.invokeMethod(service, "mergeChunks", parsedFile, chunks, allowMissing);
|
||||
}
|
||||
|
||||
@Test
|
||||
void missingChunkIndexesFindsGap() {
|
||||
assertEquals(List.of(3), missingChunkIndexes(List.of(chunk(1, 4, "A", "成功"), chunk(2, 4, "B", "成功"), chunk(4, 4, "D", "成功"))));
|
||||
}
|
||||
|
||||
@Test
|
||||
void missingChunkIndexesMiddleGapOf27171Scenario() {
|
||||
// 事故现场:4284 个分片、只缺 247 一个
|
||||
List<DeleteBrandResultFileDto> chunks = new java.util.ArrayList<>();
|
||||
for (int i = 1; i <= 4284; i++) {
|
||||
if (i == 247) {
|
||||
continue;
|
||||
}
|
||||
chunks.add(chunk(i, 4284, "B0" + i, "成功"));
|
||||
}
|
||||
assertEquals(List.of(247), missingChunkIndexes(chunks));
|
||||
}
|
||||
|
||||
@Test
|
||||
void missingChunkIndexesEmptyAndUnknownTotal() {
|
||||
assertEquals(List.of(), missingChunkIndexes(List.of()));
|
||||
DeleteBrandResultFileDto noTotal = new DeleteBrandResultFileDto();
|
||||
noTotal.setChunkIndex(1);
|
||||
assertEquals(List.of(), missingChunkIndexes(List.of(noTotal)));
|
||||
}
|
||||
|
||||
@Test
|
||||
void mergeChunksWithAllowMissingFillsPlaceholderStatus() {
|
||||
// parsed 有 2 个 ASIN,但只回传了 1 个(缺 B0DELETED)
|
||||
DeleteBrandParsedFileCacheDto parsed = parsedFile("B0FINE", "B0DELETED");
|
||||
List<DeleteBrandResultFileDto> chunks = List.of(chunk(1, 2, "B0FINE", "成功"));
|
||||
Object merged = mergeChunks(parsed, chunks, true);
|
||||
assertEquals(2, mergedAsins(merged).size());
|
||||
assertEquals("成功", mergedAsins(merged).get("B0FINE"));
|
||||
assertEquals(DeleteBrandRunService.MISSING_CHUNK_STATUS, mergedAsins(merged).get("B0DELETED"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void mergeChunksStrictStillThrowsOnMissingAsin() {
|
||||
DeleteBrandParsedFileCacheDto parsed = parsedFile("B0FINE", "B0DELETED");
|
||||
List<DeleteBrandResultFileDto> chunks = List.of(chunk(1, 2, "B0FINE", "成功"));
|
||||
BusinessException ex = assertThrows(BusinessException.class,
|
||||
() -> mergeChunks(parsed, chunks, false));
|
||||
assertTrue(ex.getMessage().contains("缺少 ASIN 结果"));
|
||||
assertTrue(ex.getMessage().contains("B0DELETED"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void mergeChunksAllowMissingWholeCountryBecomesPlaceholder() {
|
||||
DeleteBrandParsedFileCacheDto parsed = parsedFile("B0FINE", "B0DELETED");
|
||||
List<DeleteBrandResultFileDto> chunks = List.of();
|
||||
Object merged = mergeChunks(parsed, chunks, true);
|
||||
assertEquals(2, mergedAsins(merged).size());
|
||||
assertEquals(DeleteBrandRunService.MISSING_CHUNK_STATUS, mergedAsins(merged).get("B0FINE"));
|
||||
assertEquals(DeleteBrandRunService.MISSING_CHUNK_STATUS, mergedAsins(merged).get("B0DELETED"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void mergeChunksStrictThrowsOnMissingCountry() {
|
||||
DeleteBrandParsedFileCacheDto parsed = parsedFile("B0FINE");
|
||||
BusinessException ex = assertThrows(BusinessException.class,
|
||||
() -> mergeChunks(parsed, List.of(), false));
|
||||
assertTrue(ex.getMessage().contains("缺少国家结果"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void mergeChunksInconsistentChunkTotalStillThrows() {
|
||||
DeleteBrandParsedFileCacheDto parsed = parsedFile("B0FINE");
|
||||
BusinessException ex = assertThrows(BusinessException.class,
|
||||
() -> mergeChunks(parsed, List.of(chunk(1, 3, "B0FINE", "成功"), chunk(2, 4, "X", "成功")), true));
|
||||
assertTrue(ex.getMessage().contains("chunkTotal 不一致"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 把 mergeChunks 返回的 MergedDeleteBrandFile(record)转成 asin→status 映射便于断言。
|
||||
*/
|
||||
private Map<String, String> mergedAsins(Object merged) {
|
||||
LinkedHashMap<String, String> map = new LinkedHashMap<>();
|
||||
List<?> countries = (List<?>) ReflectionTestUtils.getField(merged, "countries");
|
||||
for (Object country : countries) {
|
||||
List<?> items = (List<?>) ReflectionTestUtils.getField(country, "items");
|
||||
for (Object item : items) {
|
||||
map.put((String) ReflectionTestUtils.getField(item, "asin"), (String) ReflectionTestUtils.getField(item, "status"));
|
||||
}
|
||||
}
|
||||
return map;
|
||||
}
|
||||
}
|
||||
+36
-1
@@ -8,6 +8,7 @@ import com.nanri.aiimage.config.DeleteBrandProgressProperties;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
|
||||
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
||||
import org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
@@ -26,7 +27,9 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyBoolean;
|
||||
import static org.mockito.ArgumentMatchers.anyLong;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.ArgumentMatchers.isNull;
|
||||
import static org.mockito.Mockito.lenient;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.times;
|
||||
@@ -46,6 +49,7 @@ class DeleteBrandStaleTaskServiceTest {
|
||||
@Mock private DeleteBrandRunService deleteBrandRunService;
|
||||
@Mock private DeleteBrandProgressProperties deleteBrandProgressProperties;
|
||||
@Mock private TaskDistributedLockService taskDistributedLockService;
|
||||
@Mock private TaskFileJobService taskFileJobService;
|
||||
|
||||
@BeforeAll
|
||||
static void initializeTableInfo() {
|
||||
@@ -178,17 +182,48 @@ class DeleteBrandStaleTaskServiceTest {
|
||||
verify(deleteBrandRunService, never()).tryFinalizeTask(anyLong(), anyBoolean());
|
||||
}
|
||||
|
||||
@Test
|
||||
void staleFailReasonIncludesHeartbeatAndAssembleError() {
|
||||
FileTaskEntity t1 = runningTask(801L);
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(t1));
|
||||
when(fileTaskMapper.selectBatchIds(any())).thenReturn(List.of(runningTask(801L)));
|
||||
when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1);
|
||||
when(deleteBrandTaskCacheService.getProgress(801L))
|
||||
.thenReturn(Map.of("last_heartbeat_at", System.currentTimeMillis() - 35 * 60_000L));
|
||||
lockAvailable();
|
||||
// 必须放在 lockAvailable() 之后:后设置的 stub 覆盖前面的 lenient 默认值
|
||||
when(taskFileJobService.latestFailedAssembleError(801L, "DELETE_BRAND"))
|
||||
.thenReturn("结果分片未完整: 240d70221b104a3c8140555efed4fd21,缺失分片: [247](已收 4283/4284)");
|
||||
DeleteBrandStaleTaskService service = service();
|
||||
|
||||
failStaleDeleteBrandTasks(service);
|
||||
|
||||
@SuppressWarnings({"rawtypes", "unchecked"})
|
||||
ArgumentCaptor<LambdaUpdateWrapper> update = ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
|
||||
verify(fileTaskMapper).update(isNull(), update.capture());
|
||||
// 组装后的 SQL 只含占位符,真实文案在 paramNameValuePairs 中
|
||||
List<String> values = update.getValue().getParamNameValuePairs().values().stream()
|
||||
.map(String::valueOf)
|
||||
.toList();
|
||||
String errorMessage = values.stream().filter(v -> v.contains("最后心跳")).findFirst().orElse("");
|
||||
// 文案必须包含真实原因(最后心跳 + 最近一次组装失败),不能只写"回传长时间无响应"
|
||||
assertTrue(errorMessage.contains("最后心跳"), errorMessage);
|
||||
assertTrue(errorMessage.contains("最近一次组装失败"), errorMessage);
|
||||
assertTrue(errorMessage.contains("缺失分片"), errorMessage);
|
||||
}
|
||||
|
||||
private DeleteBrandStaleTaskService service() {
|
||||
return new DeleteBrandStaleTaskService(
|
||||
fileTaskMapper, deleteBrandTaskCacheService, deleteBrandTaskStorageService, deleteBrandRunService,
|
||||
null, null, null, null, null, null, null, null, null, null, null, null, null, null, null,
|
||||
deleteBrandProgressProperties, null, taskDistributedLockService, null);
|
||||
deleteBrandProgressProperties, null, taskDistributedLockService, taskFileJobService);
|
||||
}
|
||||
|
||||
private void lockAvailable() {
|
||||
when(deleteBrandProgressProperties.getHeartbeatTimeoutMinutes()).thenReturn(15L);
|
||||
when(taskDistributedLockService.acquire(any(), any(), anyLong()))
|
||||
.thenReturn(mock(TaskDistributedLockService.LockHandle.class));
|
||||
lenient().when(taskFileJobService.latestFailedAssembleError(any(), anyString())).thenReturn(null);
|
||||
}
|
||||
|
||||
private static void failStaleDeleteBrandTasks(DeleteBrandStaleTaskService service) {
|
||||
|
||||
+11
@@ -10,6 +10,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* DeleteBrandResultFileJobHandler 测试(任务 70)。
|
||||
@@ -64,6 +65,16 @@ class DeleteBrandResultFileJobHandlerTest {
|
||||
assertFalse(handler.isOwnerScoped());
|
||||
}
|
||||
|
||||
@Test
|
||||
void fallbackAssembleDelegates() {
|
||||
TaskFileJobEntity job = new TaskFileJobEntity();
|
||||
job.setTaskId(27171L);
|
||||
job.setResultId(30195L);
|
||||
when(deleteBrandRunService.fallbackAssembleResultFile(job)).thenReturn(true);
|
||||
assertEquals(true, handler.fallbackAssembleOnFailure(job, "结果分片未完整"));
|
||||
verify(deleteBrandRunService).fallbackAssembleResultFile(job);
|
||||
}
|
||||
|
||||
@Test
|
||||
void nullJobGuarded() {
|
||||
assertTrue(handler.process(null));
|
||||
|
||||
Reference in New Issue
Block a user