修复双实例 owner-scoped 结果文件 job 被非归属实例误杀
Build Backend JAR / build (push) Has been cancelled

线上现象:货源查询任务 26402/26400(归属 server-121)在归属机正常推进,
却被 server-110 判为「结果生成失败」,错误为「该任务已绑定到另一台服务实例处理」。

两个缺陷叠加:
1. 两台实例共用一个 RocketMQ consumer group,dispatch 消息被轮询投递;
   worker 对"PENDING + owner 不匹配"的 job 选择本地抢跑,模块层
   ensureTaskOwnedByCurrentInstance 直接抛 TaskOwnerMismatchException,
   异常落进通用 catch 被当成生成失败,5 次抢跑即耗尽 retryCount 并终态失败。
2. runPendingJobs 双重 claim:claimRunnableJobsForOwner 已把 job 翻成 RUNNING,
   随后又走 claimRunning(条件 PENDING/FAILED)必然失败 → job 被静默丢弃,
   归属机的定时兜底从未真正生效。

修复:
- worker:非归属实例一律跳过 owner-scoped job;抽出 processClaimed 供已持有
  claim 的调用方使用;processInternal 单独捕获 TaskOwnerMismatchException,
  只 deferRunning 退回 PENDING,不计入重试。
- TaskFileJobLocalDispatcher 新增 dispatchClaimed。
- dispatch 事件带上 owner,coordinator 按归属路由:本机 owner 直接本地派发不发 MQ,
  他机 owner 不发不跑交由归属机轮询接手,owner 为空维持原有 MQ 行为。
- 新增 owner 路由回归测试两组,回滚配合"本地抢跑"的旧断言。

同时提交此前工作区内已随 JAR 上线的 backend-java 改动:SimilarAsin 解析载荷
groups 内嵌 items(Python 旧链路兼容)与 chunk 类型不匹配跳过、
PermissionMenuSchemaInitializer 与 V103 两级菜单分组迁移、相关测试。
This commit is contained in:
2026-09-01 21:07:25 +08:00
parent 55517f996b
commit 5f4fcad2ef
16 changed files with 679 additions and 44 deletions
@@ -106,8 +106,17 @@ class ExplainIndexAuditDocTest {
assertTrue(audit.contains("不做 DDL") || audit.contains("本任务无 DDL")
|| audit.contains("不新增索引") || audit.contains("只审计"),
"审计文档明确本任务不做 DDL");
assertFalse(audit.contains("ALTER TABLE"),
"审计文档不包含 ALTER TABLEDDL 属于后续任务 116");
// 任务 116 落地 V101 索引后,审计文档会把 ALTER TABLE 作为候选索引的
// 回滚说明引用。审计任务自身依然不执行 DDL——判定口径改为「ALTER TABLE
// 只能出现在候选/回滚等说明性上下文」,而不是文档里不得出现该字符串。
for (String line : audit.split("\r?\n")) {
if (!line.contains("ALTER TABLE")) {
continue;
}
assertTrue(line.contains("回滚") || line.contains("候选") || line.contains("建议")
|| line.contains("风险") || line.trim().startsWith(">"),
"审计文档中的 ALTER TABLE 必须是候选/回滚说明,实际行:" + line);
}
assertTrue(spec.contains("只读") || spec.contains("巡检"),
"12 spec 巡检报表模式与审计文档对应");
}
@@ -27,6 +27,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -268,6 +269,7 @@ class ExternalCallMetricsRecorderTest {
}
} finally {
pool.shutdown();
assertTrue(pool.awaitTermination(60, TimeUnit.SECONDS), "并发任务在超时前全部结束");
}
awaitMetric("aiimage.external-call.total", "client", "llm", "result", "success");
@@ -275,6 +277,12 @@ class ExternalCallMetricsRecorderTest {
Thread.sleep(10);
}
assertEquals(20, llmSubmitCount.get(), "并发 20 请求全部真实发出");
// 指标在响应处理完成之后才落,submit 计数到 20 不代表指标已收敛:
// 直接断言会读到中间值(实测 18.0)。等指标计数自身收敛再断言。
for (int i = 0; i < 2000 && counterCount("aiimage.external-call.total",
"client", "llm", "result", "success") < 20.0; i++) {
Thread.sleep(10);
}
assertEquals(20.0, counterCount("aiimage.external-call.total",
"client", "llm", "result", "success"), "20 次成功全部记录,无重复");
}
@@ -7,6 +7,7 @@ import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
import org.apache.poi.ss.usermodel.Sheet;
import org.apache.poi.ss.usermodel.Workbook;
import org.apache.poi.ss.usermodel.WorkbookFactory;
import org.apache.poi.util.DefaultTempFileCreationStrategy;
import org.apache.poi.util.TempFile;
import org.apache.poi.util.TempFileCreationStrategy;
import org.junit.jupiter.api.AfterEach;
@@ -39,20 +40,36 @@ class CollectDataExcelCleanupTest {
private final List<File> trackedTempFiles = new ArrayList<>();
private Path sxssfTempDir;
@AfterEach
void restoreDefaultTempFileStrategy() {
// 还原 POI 的进程级全局策略:trackSxssfTempFiles 改写的是单例,
// 不还原会泄漏到同一 JVM 内后续所有用到 SXSSF 的测试。
// (原来这个 @AfterEach 挂在 private 的 trackSxssfTempFiles 上,
// 既没有还原策略,方法本身还是被各用例手工调用的 setup 辅助。)
TempFile.setTempFileCreationStrategy(new DefaultTempFileCreationStrategy());
}
private void trackSxssfTempFiles() {
// 把 SXSSF 滚动窗口临时文件重定向到本用例独占目录,使残留探测只覆盖
// 本用例自己产生的文件。
try {
sxssfTempDir = Files.createDirectory(tempDir.resolve("sxssf-probe"));
} catch (IOException ex) {
throw new IllegalStateException(ex);
}
TempFile.setTempFileCreationStrategy(new TempFileCreationStrategy() {
@Override
public File createTempFile(String prefix, String suffix) throws IOException {
File file = File.createTempFile(prefix, suffix);
File file = Files.createTempFile(sxssfTempDir, prefix, suffix).toFile();
trackedTempFiles.add(file);
return file;
}
@Override
public File createTempDirectory(String prefix) throws IOException {
File dir = Files.createTempDirectory(prefix).toFile();
File dir = Files.createTempDirectory(sxssfTempDir, prefix).toFile();
trackedTempFiles.add(dir);
return dir;
}
@@ -62,10 +79,13 @@ class CollectDataExcelCleanupTest {
private void assertSxssfTempFilesCleaned() {
assertThat(trackedTempFiles).as("SXSSF 滚动窗口临时文件均已删除")
.allSatisfy(f -> assertThat(f).doesNotExist());
// POI 5.2.5 无法读取当前策略,改用系统临时目录中 SXSSF 固定前缀的残留探测。
try (Stream<Path> paths = Files.list(Path.of(System.getProperty("java.io.tmpdir")))) {
// 只探测本用例独占目录。原实现扫描系统共享临时目录并要求 poi-sxssf*
// 前缀零命中,会把其他进程、其他用例和历史运行遗留的文件算成本用例
// 的泄漏——本机共享临时目录里就躺着十几个历史 poi-sxssf-sheet*.xml
// 导致实现完全正确时这 7 个用例照样红。
try (Stream<Path> paths = Files.list(sxssfTempDir)) {
assertThat(paths.filter(p -> p.getFileName().toString().startsWith("poi-sxssf")))
.as("系统临时目录无 SXSSF 滚动窗口残留")
.as("本用例 SXSSF 临时目录无滚动窗口残留")
.isEmpty();
} catch (IOException ex) {
throw new IllegalStateException(ex);
@@ -149,6 +149,7 @@ class ShopDataCrawlCleanupTest {
private final AtomicLong memberIdSeq = new AtomicLong(1000);
private final AtomicLong fileSeq = new AtomicLong(5000);
private Long lastJobTaskId;
private Set<Path> preExistingWorkRoots = Set.of();
@BeforeEach
void configureStorage() {
@@ -408,6 +409,12 @@ class ShopDataCrawlCleanupTest {
.deleteTaskItems(anyLong(), eq(MODULE_TYPE));
lenient().doNothing().when(taskProgressSnapshotService)
.delete(anyLong(), eq(MODULE_TYPE));
// 共享临时目录里的既有 daily-* 目录不属于本用例,先快照、断言时做差集
try {
preExistingWorkRoots = Set.copyOf(scanWorkRoots());
} catch (Exception ex) {
throw new IllegalStateException(ex);
}
}
// ---- 1. 删除路径 ----
@@ -1261,6 +1268,16 @@ class ShopDataCrawlCleanupTest {
}
private List<Path> workRoots() throws Exception {
// 只统计本用例新产生的工作目录。shop-data-crawl-result 落在机器共享的
// 系统临时目录下,其他用例、历史运行、本机跑过的应用都会往里留 daily-*
// 目录(本机实测已堆了 200+ 个),原实现把这些全算作本用例的泄漏,
// 于是实现完全正确时这些用例照样红。改为与用例开始时的快照做差集。
List<Path> current = new ArrayList<>(scanWorkRoots());
current.removeAll(preExistingWorkRoots);
return current;
}
private List<Path> scanWorkRoots() throws Exception {
Path root = Path.of(System.getProperty("java.io.tmpdir"), "shop-data-crawl-result");
if (!Files.exists(root)) {
return List.of();
@@ -336,6 +336,56 @@ class SimilarAsinTaskServiceGroupRefTest {
assertEquals(80, coverage);
}
@Test
void test_task_006_group_payload_embeds_items_for_python() throws Exception {
// Python 旧链路(normalize_groups 直接返回 groups 并按 gp["items"] 取行):
// 写入存储的 payload 中每组必须内嵌完整行,startIndex/endIndex 保留供 Java 内部展开。
java.util.concurrent.atomic.AtomicReference<String> captured = new java.util.concurrent.atomic.AtomicReference<>();
when(transientPayloadStorageService.storeParsedPayloadFast(
eq(SimilarAsinTaskService.MODULE_TYPE), any(), anyString(), anyString(), eq(false)))
.thenAnswer(invocation -> {
captured.set(invocation.getArgument(3));
return "rustfs:task-parsed/similar-asin/30000/payload.json";
});
File workbook = buildWorkbook(12);
SimilarAsinParseVo vo = parse(workbook, "uploads/20260829/gr-python.xlsx");
assertEquals(12, vo.getAcceptedRows());
SimilarAsinParsedPayloadDto stored = objectMapper.readValue(captured.get(), SimilarAsinParsedPayloadDto.class);
assertNotNull(stored.getGroups());
assertEquals(12, stored.getGroups().size());
// 每组内嵌 items 与引用区间宽度一致,且行内容对应 items[startIndex]
for (SimilarAsinParsedGroupVo group : stored.getGroups()) {
assertNotNull(group.getItems());
assertEquals(group.getEndIndex() - group.getStartIndex(), group.getItems().size(),
"Python 兼容:组内嵌 items 必须与引用区间宽度一致");
int start = group.getStartIndex();
assertEquals(stored.getItems().get(start).getRowToken(), group.getItems().get(0).getRowToken(),
"组 items 首行必须是 items[startIndex]");
}
// Java 内部展开路径不受内嵌 items 影响:仍按引用切片
List<SimilarAsinParsedRowVo> expanded = SimilarAsinTaskService.expandGroupRefs(stored);
assertEquals(12, expanded.size(), "Java 内部展开仍按引用,不受内嵌 items 影响");
}
@Test
void test_task_006_group_hydrate_embeds_items_for_legacy_payload() throws Exception {
// 存量任务 payload 只有 startIndex/endIndex 引用(无内嵌 items):
// parsedPayload 读取时 hydrate 必须回填组内嵌 itemsPython 才能取到行。
String legacy = "{\"items\":["
+ "{\"rowToken\":\"t1\",\"asin\":\"B0GRP00001\",\"sourceFileKey\":\"f.xlsx\",\"country\":\"英国\"},"
+ "{\"rowToken\":\"t2\",\"asin\":\"B0GRP00002\",\"sourceFileKey\":\"f.xlsx\",\"country\":\"英国\"}"
+ "],\"groups\":[{\"groupKey\":\"g1\",\"startIndex\":0,\"endIndex\":2,\"itemCount\":2}]}";
SimilarAsinParsedPayloadDto payload = objectMapper.readValue(legacy, SimilarAsinParsedPayloadDto.class);
SimilarAsinParsedPayloadDto hydrated = SimilarAsinTaskService.hydrateForPythonCompat(payload);
assertEquals(1, hydrated.getGroups().size());
assertNotNull(hydrated.getGroups().get(0).getItems());
assertEquals(2, hydrated.getGroups().get(0).getItems().size(), "hydrate 后组内嵌 items 必须为 2 行");
assertEquals("B0GRP00001", hydrated.getGroups().get(0).getItems().get(0).getAsin(),
"hydrate 后组内嵌 items 首行必须对应 items[0]");
assertEquals("B0GRP00002", hydrated.getGroups().get(0).getItems().get(1).getAsin(),
"hydrate 后组内嵌 items 第二行必须对应 items[1]");
}
// ---- helpers ----
private static String groupRefJson(int groupCount, int[][] ranges) {
@@ -0,0 +1,85 @@
package com.nanri.aiimage.modules.task.service;
import com.nanri.aiimage.config.InstanceMetadata;
import com.nanri.aiimage.modules.task.model.dto.TaskFileJobDispatchEvent;
import com.nanri.aiimage.modules.task.model.dto.TaskFileJobMessage;
import org.junit.jupiter.api.Test;
import org.springframework.core.task.TaskExecutor;
import java.time.LocalDateTime;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
/**
* owner-scoped job 的派发路由(2026-09-01 事故):两台实例共用一个 RocketMQ consumer group
* 消息被轮询投递到非归属实例后只能丢弃。归属实例改为直接本地派发,非归属实例不再发消息。
*/
class TaskFileJobDispatchCoordinatorOwnerRoutingTest {
private final TaskFileJobPublisher publisher = mock(TaskFileJobPublisher.class);
private final TaskFileJobLocalDispatcher localDispatcher = mock(TaskFileJobLocalDispatcher.class);
private final InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
private final TaskExecutor inlineExecutor = Runnable::run;
private TaskFileJobDispatchCoordinator buildCoordinator() {
return new TaskFileJobDispatchCoordinator(publisher, localDispatcher, instanceMetadata, inlineExecutor);
}
private static TaskFileJobDispatchEvent event(String owner) {
return new TaskFileJobDispatchEvent(18524L, 26402L, "SIMILAR_ASIN", 29405L,
owner == null ? "task:26402" : "task:26402:owner:" + owner,
TaskFileJobService.JOB_TYPE_ASSEMBLE_RESULT, owner, LocalDateTime.now());
}
@Test
void ownerJobDispatchedLocallyWithoutMq() {
when(instanceMetadata.getInstanceId()).thenReturn("server-121");
when(localDispatcher.dispatch(18524L, 26402L, "SIMILAR_ASIN", true)).thenReturn(true);
buildCoordinator().onTaskFileJobDispatch(event("server-121"));
verify(localDispatcher).dispatch(18524L, 26402L, "SIMILAR_ASIN", true);
verifyNoInteractions(publisher);
}
@Test
void ownerJobFallsBackToMqWhenLocalDispatchDeclined() {
when(instanceMetadata.getInstanceId()).thenReturn("server-121");
when(localDispatcher.dispatch(18524L, 26402L, "SIMILAR_ASIN", true)).thenReturn(false, true);
when(publisher.publish(any(TaskFileJobMessage.class))).thenReturn(false);
buildCoordinator().onTaskFileJobDispatch(event("server-121"));
verify(publisher).publish(any(TaskFileJobMessage.class));
}
@Test
void otherInstanceJobIsNeitherPublishedNorRun() {
when(instanceMetadata.getInstanceId()).thenReturn("server-110");
buildCoordinator().onTaskFileJobDispatch(event("server-121"));
verifyNoInteractions(publisher);
verify(localDispatcher, never()).dispatch(anyLong(), anyLong(), any(), anyBoolean());
verify(localDispatcher, never()).dispatch(anyLong(), anyLong(), any());
}
@Test
void jobWithoutOwnerStillGoesThroughMq() {
when(instanceMetadata.getInstanceId()).thenReturn("server-110");
when(publisher.publish(any(TaskFileJobMessage.class))).thenReturn(true);
buildCoordinator().onTaskFileJobDispatch(event(null));
verify(publisher).publish(any(TaskFileJobMessage.class));
verify(localDispatcher, never()).dispatch(anyLong(), anyLong(), eq("SIMILAR_ASIN"), anyBoolean());
}
}
@@ -0,0 +1,163 @@
package com.nanri.aiimage.modules.task.service;
import com.nanri.aiimage.common.exception.TaskOwnerMismatchException;
import com.nanri.aiimage.config.InstanceMetadata;
import com.nanri.aiimage.modules.appearancepatent.service.AppearancePatentTaskService;
import com.nanri.aiimage.modules.brand.service.BrandTaskService;
import com.nanri.aiimage.modules.collectdata.service.CollectDataService;
import com.nanri.aiimage.modules.deletebrand.service.DeleteBrandRunService;
import com.nanri.aiimage.modules.patroldelete.service.PatrolDeleteTaskService;
import com.nanri.aiimage.modules.pricetrack.service.PriceTrackTaskService;
import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskService;
import com.nanri.aiimage.modules.publish.service.PublishTaskService;
import com.nanri.aiimage.modules.queryasin.service.QueryAsinTaskService;
import com.nanri.aiimage.modules.shopdatacrawl.service.ShopDataCrawlTaskService;
import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskService;
import com.nanri.aiimage.modules.similarasin.service.SimilarAsinTaskService;
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskService;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Field;
import java.time.LocalDateTime;
import java.util.List;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* 双实例归属路由回归测试(2026-09-01 货源查询任务被非归属实例误杀):
* 1) 定时 worker 已 claim 的 job 不能再走 claimRunning,否则条件 UPDATE 失败被静默丢弃;
* 2) 处理中撞上归属校验时只退回 PENDING,不得计入 retryCountmarkFailed)。
*/
class TaskResultFileJobWorkerOwnerRoutingTest {
private final ShopMatchTaskService shopMatch = mock(ShopMatchTaskService.class);
private final PriceTrackTaskService priceTrack = mock(PriceTrackTaskService.class);
private final ProductRiskTaskService productRisk = mock(ProductRiskTaskService.class);
private final PublishTaskService publish = mock(PublishTaskService.class);
private final QueryAsinTaskService queryAsin = mock(QueryAsinTaskService.class);
private final ShopDataCrawlTaskService shopDataCrawl = mock(ShopDataCrawlTaskService.class);
private final WithdrawTaskService withdraw = mock(WithdrawTaskService.class);
private final PatrolDeleteTaskService patrolDelete = mock(PatrolDeleteTaskService.class);
private final AppearancePatentTaskService appearance = mock(AppearancePatentTaskService.class);
private final SimilarAsinTaskService similar = mock(SimilarAsinTaskService.class);
private final DeleteBrandRunService deleteBrand = mock(DeleteBrandRunService.class);
private final BrandTaskService brand = mock(BrandTaskService.class);
private final CollectDataService collectData = mock(CollectDataService.class);
private final TaskResultPayloadService payload = mock(TaskResultPayloadService.class);
private final TaskFileJobService taskFileJobService = mock(TaskFileJobService.class);
private final TaskDistributedLockService taskDistributedLockService = mock(TaskDistributedLockService.class);
private final TaskDistributedLockService.LockHandle lock = mock(TaskDistributedLockService.LockHandle.class);
private final TaskFileJobLocalDispatcher localDispatcher = mock(TaskFileJobLocalDispatcher.class);
private final InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
private TaskResultFileJobWorker buildWorker() throws Exception {
List<ResultFileJobHandler> handlers = List.of(
new ShopMatchResultFileJobHandler(shopMatch, payload),
new PriceTrackResultFileJobHandler(priceTrack, payload),
new ProductRiskResultFileJobHandler(productRisk, payload),
new PublishResultFileJobHandler(publish),
new QueryAsinResultFileJobHandler(queryAsin, payload),
new ShopDataCrawlResultFileJobHandler(shopDataCrawl, payload),
new WithdrawResultFileJobHandler(withdraw, payload),
new PatrolDeleteResultFileJobHandler(patrolDelete, payload),
new AppearancePatentResultFileJobHandler(appearance),
new SimilarAsinResultFileJobHandler(similar),
new DeleteBrandResultFileJobHandler(deleteBrand),
new BrandResultFileJobHandler(brand, payload),
new CollectDataResultFileJobHandler(collectData));
TaskResultFileJobWorker worker = new TaskResultFileJobWorker(
taskFileJobService,
taskDistributedLockService,
mock(FileResultMapper.class),
localDispatcher,
instanceMetadata,
withdraw, brand,
new ResultFileJobHandlerRegistry(handlers));
set(worker, "taskQueueExecutor", (org.springframework.core.task.TaskExecutor) Runnable::run);
set(worker, "localWorkerEnabled", true);
set(worker, "batchSize", 20);
return worker;
}
private static void set(TaskResultFileJobWorker worker, String fieldName, Object value) throws Exception {
Field field = TaskResultFileJobWorker.class.getDeclaredField(fieldName);
field.setAccessible(true);
field.set(worker, value);
}
private static TaskFileJobEntity claimedJob(String moduleType, long jobId, long taskId, String scopeKey) {
TaskFileJobEntity entity = new TaskFileJobEntity();
entity.setId(jobId);
entity.setTaskId(taskId);
entity.setResultId(taskId + 1000L);
entity.setModuleType(moduleType);
entity.setScopeKey(scopeKey);
entity.setStatus("RUNNING");
entity.setUpdatedAt(LocalDateTime.now());
return entity;
}
@Test
void scheduledWorkerProcessesAlreadyClaimedJobWithoutReclaiming() throws Exception {
TaskResultFileJobWorker worker = buildWorker();
when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
TaskFileJobEntity claim = claimedJob(PublishTaskService.MODULE_TYPE, 18524L, 26402L, "task:26402:owner:instance-a");
when(taskFileJobService.claimRunnableJobsForOwner(20, "instance-a")).thenReturn(List.of(claim));
when(taskFileJobService.activateRunningClaim(claim)).thenReturn(true);
when(localDispatcher.dispatchClaimed(claim.getId(), claim.getTaskId(), claim.getModuleType())).thenReturn(false);
when(taskDistributedLockService.acquire(PublishTaskService.MODULE_TYPE, claim.getTaskId(),
TaskDistributedLockService.DEFAULT_WAIT_MILLIS)).thenReturn(lock);
worker.runPendingJobs();
verify(taskFileJobService, never()).claimRunning(anyLong());
verify(publish).processResultFileJob(claim);
verify(taskFileJobService).markSuccess(eq(claim), any());
}
@Test
void scheduledWorkerPrefersClaimedDispatch() throws Exception {
TaskResultFileJobWorker worker = buildWorker();
when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
TaskFileJobEntity claim = claimedJob(PublishTaskService.MODULE_TYPE, 18525L, 26403L, "task:26403:owner:instance-a");
when(taskFileJobService.claimRunnableJobsForOwner(20, "instance-a")).thenReturn(List.of(claim));
when(localDispatcher.dispatchClaimed(claim.getId(), claim.getTaskId(), claim.getModuleType())).thenReturn(true);
worker.runPendingJobs();
verify(localDispatcher).dispatchClaimed(claim.getId(), claim.getTaskId(), claim.getModuleType());
verify(localDispatcher, never()).dispatch(anyLong(), anyLong(), any());
verify(taskFileJobService, never()).claimRunning(anyLong());
verify(publish, never()).processResultFileJob(any());
}
@Test
void ownerMismatchDuringProcessingDefersInsteadOfFailing() throws Exception {
TaskResultFileJobWorker worker = buildWorker();
when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
// scopeKey 归属本机(通过 process 的前置校验),但模块层读到的 task owner 已经变成他机
TaskFileJobEntity job = claimedJob(PublishTaskService.MODULE_TYPE, 18526L, 26404L, "task:26404:owner:instance-a");
when(taskFileJobService.claimRunning(job.getId())).thenReturn(job);
when(taskFileJobService.activateRunningClaim(job)).thenReturn(true);
when(taskDistributedLockService.acquire(PublishTaskService.MODULE_TYPE, job.getTaskId(),
TaskDistributedLockService.DEFAULT_WAIT_MILLIS)).thenReturn(lock);
doThrow(new TaskOwnerMismatchException(job.getTaskId(), "assemble result file", "instance-b", "instance-a"))
.when(publish).processResultFileJob(job);
worker.process(job);
verify(taskFileJobService).deferRunning(eq(job.getId()), any());
verify(taskFileJobService, never()).markFailed(any(), any());
verify(taskFileJobService, never()).markFailedPermanent(any(), any());
verify(taskFileJobService, never()).isRetryExhausted(anyLong());
}
}
@@ -128,6 +128,20 @@ class TaskResultFileJobWorkerOwnerScopedTest {
TaskResultFileJobWorker worker = buildWorker();
when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
TaskFileJobEntity job = job("PUBLISH", 2L, 12L, "task:12:owner:instance-b");
job.setStatus("RUNNING");
worker.process(job);
verifyNoInteractions(taskFileJobService, taskDistributedLockService, publish);
}
@Test
void ownerScopedOtherOwnerPendingSkip() {
// PENDING + owner 不匹配也必须跳过:模块层 ensureTaskOwnedByCurrentInstance 会抛
// TaskOwnerMismatchException,本机抢跑只会耗尽 job 重试次数并误杀归属机上的任务。
// job 由归属实例的 runPendingJobs(每 15 秒扫自己 owner 的 PENDING/FAILED)接手。
TaskResultFileJobWorker worker = buildWorker();
when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
TaskFileJobEntity job = job("PUBLISH", 2L, 12L, "task:12:owner:instance-b");
job.setStatus("PENDING");
worker.process(job);
verifyNoInteractions(taskFileJobService, taskDistributedLockService, publish);
}