super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

处理数据量binlog问题

Browse Source
This commit is contained in:
supernijia
2026-07-28 15:49:41 +08:00
parent 7c3c9e53a0
commit 9048bbb7f8
8 changed files with 187 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java
+2
View File
@@ -33,6 +33,8 @@ public class SimilarAsinProperties {
private int cozeReadTimeoutMillis = 60000;
private int cozePollIntervalMillis = 30000;
private int cozePollTimeoutMillis = 1800000;
private long dbTaskTouchIntervalMillis = 120000L;
private long dbJobTouchIntervalMillis = 60000L;
private int staleTimeoutMinutes = 30;
private String staleFinalizeCron = "0 */2 * * * *";
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java
+21 -15
View File
@@ -1026,7 +1026,7 @@ public class SimilarAsinTaskService {
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
boolean pendingCoze = submitCozeBatches(task, result, job, chunks, allRowsByBaseId);
if (pendingCoze) {
taskFileJobService.touchRunning(job.getId());
taskFileJobService.touchRunningIfStale(job.getId(), properties.getDbJobTouchIntervalMillis());
touchJavaSideTaskActivity(task.getId());
} else if (isResultSubmissionComplete(task.getId())) {
maybeFinalizeCozeJobLocked(task.getId(), new CozeBatchContext(
@@ -1715,7 +1715,7 @@ public class SimilarAsinTaskService {
finalizeTimedOutCozeStatesForTask(task.getId());
}
if (countPendingCozeStates(task.getId()) > 0) {
taskFileJobService.touchRunning(job.getId());
taskFileJobService.touchRunningIfStale(job.getId(), properties.getDbJobTouchIntervalMillis());
touchJavaSideTaskActivity(task.getId());
saveFileBuildProgress(task, job, totalProgressUnits, 1, "Coze 已提交，等待结果");
return false;
@@ -1737,13 +1737,13 @@ public class SimilarAsinTaskService {
saveFileBuildProgress(task, job, totalProgressUnits, 0, "正在提交 Coze");
boolean pendingCoze = submitCozeBatches(task, result, job, chunks, allRowsByBaseId);
if (pendingCoze) {
taskFileJobService.touchRunning(job.getId());
taskFileJobService.touchRunningIfStale(job.getId(), properties.getDbJobTouchIntervalMillis());
touchJavaSideTaskActivity(task.getId());
saveFileBuildProgress(task, job, totalProgressUnits, 1, "Coze 已提交，等待结果");
return false;
}
if (STATUS_RUNNING.equals(task.getStatus()) && !isResultSubmissionComplete(task.getId())) {
taskFileJobService.touchRunning(job.getId());
taskFileJobService.touchRunningIfStale(job.getId(), properties.getDbJobTouchIntervalMillis());
touchJavaSideTaskActivity(task.getId());
saveFileBuildProgress(task, job, totalProgressUnits, Math.max(1, cozeWorkUnits),
"等待 Python 上传，每 " + batchSize + " 行提交一次 Coze");
@@ -2140,7 +2140,7 @@ public class SimilarAsinTaskService {
if (batchRows.isEmpty()) {
return false;
}
taskFileJobService.touchRunning(job.getId());
taskFileJobService.touchRunningIfStale(job.getId(), properties.getDbJobTouchIntervalMillis());
String batchScopeKey = buildCozeBatchScopeKey(task.getId(), batchRows);
String batchScopeHash = DigestUtil.sha256Hex(batchScopeKey);
TaskScopeStateEntity existing = taskScopeStateMapper.selectOne(new LambdaQueryWrapper<TaskScopeStateEntity>()
@@ -2188,7 +2188,7 @@ public class SimilarAsinTaskService {
if (CozeFailureClassifier.isThrottleLockTimeout(message)) {
savePendingCozeBatchState(task, result, job, batchRows, batchScopeKey, batchScopeHash,
batchIndex, batchTotal, message, credential.name());
taskFileJobService.touchRunning(job.getId());
taskFileJobService.touchRunningIfStale(job.getId(), properties.getDbJobTouchIntervalMillis());
touchJavaSideTaskActivity(task.getId());
return true;
}
@@ -2359,12 +2359,11 @@ public class SimilarAsinTaskService {
state.getTaskId(), state.getId(), context.ownerInstanceId(), currentInstanceId());
return;
}
taskFileJobService.touchRunning(context.jobId());
taskFileJobService.touchRunningIfStale(context.jobId(), properties.getDbJobTouchIntervalMillis());
try {
// P0-2 最小风险变体：优先复用预取阶段的 HTTP 响应；未命中（单 state 或失败兜底）走原同步 HTTP。
SimilarAsinCozeClient.CozePollResponse poll = pollWorkflowCached(state, context.credentialName());
if (!poll.hasPayload() && !poll.isFinished() && !isCozeStateTimedOut(state)) {
updateCozeStateRunning(state, null);
return;
}
String failureMessage = poll.isFailed()
@@ -2552,7 +2551,7 @@ public class SimilarAsinTaskService {
.set(TaskScopeStateEntity::getCompleted, 0)
.set(TaskScopeStateEntity::getUpdatedAt, now));
if (updated > 0) {
taskFileJobService.touchRunning(context.jobId());
taskFileJobService.touchRunningIfStale(context.jobId(), properties.getDbJobTouchIntervalMillis());
touchJavaSideTaskActivity(state.getTaskId());
log.info("[similar-asin] coze retry submitted taskId={} stateId={} oldExecuteId={} newExecuteId={} chunk={} batch={}/{} retry={}/{} failure={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(), submit.executeId(),
@@ -2568,7 +2567,7 @@ public class SimilarAsinTaskService {
log.warn("[similar-asin] coze retry deferred by throttle lock taskId={} stateId={} executeId={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId());
deferStateForResubmit(state, msg);
taskFileJobService.touchRunning(context.jobId());
taskFileJobService.touchRunningIfStale(context.jobId(), properties.getDbJobTouchIntervalMillis());
touchJavaSideTaskActivity(state.getTaskId());
return true;
}
@@ -2824,7 +2823,7 @@ public class SimilarAsinTaskService {
}
if (submittedAny) {
markCozeStateTerminal(state, COZE_STATUS_DONE, "split retry submitted after failure: " + firstNonBlank(failureMessage, "unknown"));
taskFileJobService.touchRunning(context.jobId());
taskFileJobService.touchRunningIfStale(context.jobId(), properties.getDbJobTouchIntervalMillis());
touchJavaSideTaskActivity(state.getTaskId());
maybeFinalizeCozeJob(state.getTaskId(), context);
log.info("[similar-asin] coze split retry submitted taskId={} stateId={} chunk={} batch={}/{} parts={} retry={}/{} poisonRow={} failure={}",
@@ -2840,7 +2839,7 @@ public class SimilarAsinTaskService {
log.warn("[similar-asin] coze split retry deferred by throttle lock taskId={} stateId={} parts={}",
state.getTaskId(), state.getId(), partitions.size());
deferStateForResubmit(state, msg);
taskFileJobService.touchRunning(context.jobId());
taskFileJobService.touchRunningIfStale(context.jobId(), properties.getDbJobTouchIntervalMillis());
touchJavaSideTaskActivity(state.getTaskId());
return true;
}
@@ -2940,7 +2939,7 @@ public class SimilarAsinTaskService {
return;
}
try {
taskFileJobService.touchRunning(context.jobId());
taskFileJobService.touchRunningIfStale(context.jobId(), properties.getDbJobTouchIntervalMillis());
SimilarAsinCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task), readImgSwitch(task),
cozeClient.credentialByName(context.credentialName()), false);
@@ -3246,6 +3245,8 @@ public class SimilarAsinTaskService {
.le(TaskScopeStateEntity::getCozeLastPolledAt, now.minus(Duration.ofMillis(intervalMillis))))
.set(TaskScopeStateEntity::getCozeStatus, COZE_STATUS_RUNNING)
.set(TaskScopeStateEntity::getCozeLastPolledAt, now)
.set(TaskScopeStateEntity::getCozeAttemptCount, cozeAttemptCount(state) + 1)
.set(TaskScopeStateEntity::getCozeError, null)
.set(TaskScopeStateEntity::getUpdatedAt, now)) > 0;
}
@@ -3304,7 +3305,7 @@ public class SimilarAsinTaskService {
}
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task != null && STATUS_RUNNING.equals(task.getStatus()) && !isResultSubmissionComplete(taskId)) {
taskFileJobService.touchRunning(job.getId());
taskFileJobService.touchRunningIfStale(job.getId(), properties.getDbJobTouchIntervalMillis());
touchJavaSideTaskActivity(taskId);
return;
}
@@ -3812,11 +3813,16 @@ public class SimilarAsinTaskService {
if (taskId == null || taskId <= 0) {
return;
}
LocalDateTime now = LocalDateTime.now();
LocalDateTime cutoff = now.minus(Duration.ofMillis(Math.max(1_000L, properties.getDbTaskTouchIntervalMillis())));
fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getId, taskId)
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
.eq(FileTaskEntity::getStatus, STATUS_RUNNING)
.set(FileTaskEntity::getUpdatedAt, LocalDateTime.now()));
.and(wrapper -> wrapper.isNull(FileTaskEntity::getUpdatedAt)
.or()
.le(FileTaskEntity::getUpdatedAt, cutoff))
.set(FileTaskEntity::getUpdatedAt, now));
}
private String buildCozeBatchScopeKey(Long taskId, List<SimilarAsinResultRowDto> batchRows) {
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java
+16
View File
@@ -11,6 +11,7 @@ import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.LinkedHashMap;
@@ -212,6 +213,21 @@ public class TaskFileJobService {
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now()));
}
public void touchRunningIfStale(Long jobId, long intervalMillis) {
if (jobId == null || jobId <= 0) {
return;
}
LocalDateTime now = LocalDateTime.now();
LocalDateTime cutoff = now.minus(Duration.ofMillis(Math.max(1_000L, intervalMillis)));
taskFileJobMapper.update(null, new LambdaUpdateWrapper<TaskFileJobEntity>()
.eq(TaskFileJobEntity::getId, jobId)
.eq(TaskFileJobEntity::getStatus, "RUNNING")
.and(wrapper -> wrapper.isNull(TaskFileJobEntity::getUpdatedAt)
.or()
.le(TaskFileJobEntity::getUpdatedAt, cutoff))
.set(TaskFileJobEntity::getUpdatedAt, now));
}
public boolean deferRunning(Long jobId, String message) {
if (jobId == null || jobId <= 0) {
return false;
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java
+39
View File
@@ -2,6 +2,7 @@ package com.nanri.aiimage.modules.task.service;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.nanri.aiimage.config.SimilarAsinProperties;
import com.nanri.aiimage.modules.appearancepatent.service.AppearancePatentTaskCacheService;
import com.nanri.aiimage.modules.brand.mapper.BrandCrawlTaskMapper;
import com.nanri.aiimage.modules.brand.model.entity.BrandCrawlTaskEntity;
@@ -25,6 +26,7 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -62,6 +64,7 @@ public class TaskHeartbeatService {
private final WithdrawTaskCacheService withdrawTaskCacheService;
private final AppearancePatentTaskCacheService appearancePatentTaskCacheService;
private final SimilarAsinTaskCacheService similarAsinTaskCacheService;
private final SimilarAsinProperties similarAsinProperties;
private final DeleteBrandTaskCacheService deleteBrandTaskCacheService;
private final BrandTaskProgressCacheService brandTaskProgressCacheService;
@@ -120,6 +123,9 @@ public class TaskHeartbeatService {
task.getId(), task.getUserId(), moduleType, status);
return TaskHeartbeatVo.notAlive(moduleType, status, "task is not running");
}
if (MODULE_SIMILAR_ASIN.equals(moduleType)) {
return touchSimilarAsinHeartbeat(task);
}
LocalDateTime now = LocalDateTime.now();
int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getId, task.getId())
@@ -138,6 +144,39 @@ public class TaskHeartbeatService {
return TaskHeartbeatVo.alive(moduleType, STATUS_RUNNING);
}
private TaskHeartbeatVo touchSimilarAsinHeartbeat(FileTaskEntity task) {
LocalDateTime now = LocalDateTime.now();
long intervalMillis = Math.max(1_000L, similarAsinProperties.getDbTaskTouchIntervalMillis());
LocalDateTime cutoff = now.minus(Duration.ofMillis(intervalMillis));
boolean checkpointDue = task.getUpdatedAt() == null || !task.getUpdatedAt().isAfter(cutoff);
similarAsinTaskCacheService.touchTaskHeartbeat(task.getId());
if (!checkpointDue) {
saveFileTaskCache(MODULE_SIMILAR_ASIN, task);
return TaskHeartbeatVo.alive(MODULE_SIMILAR_ASIN, STATUS_RUNNING);
}
int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getId, task.getId())
.eq(FileTaskEntity::getStatus, STATUS_RUNNING)
.and(wrapper -> wrapper.isNull(FileTaskEntity::getUpdatedAt)
.or()
.le(FileTaskEntity::getUpdatedAt, cutoff))
.set(FileTaskEntity::getUpdatedAt, now));
if (updated <= 0) {
FileTaskEntity latest = fileTaskMapper.selectById(task.getId());
if (latest == null || !STATUS_RUNNING.equals(latest.getStatus())) {
return TaskHeartbeatVo.notAlive(MODULE_SIMILAR_ASIN,
latest == null ? task.getStatus() : latest.getStatus(), "task is not running");
}
task = latest;
} else {
task.setUpdatedAt(now);
}
saveFileTaskCache(MODULE_SIMILAR_ASIN, task);
return TaskHeartbeatVo.alive(MODULE_SIMILAR_ASIN, STATUS_RUNNING);
}
private TaskHeartbeatVo touchBrandTaskIfRunning(BrandCrawlTaskEntity task, TaskHeartbeatRequest request) {
if (task == null) {
return null;
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotService.java
+24
View File
@@ -12,6 +12,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.util.Objects;
@Service
@RequiredArgsConstructor
@@ -61,6 +62,10 @@ public class TaskProgressSnapshotService {
if (existing == null) {
throw new BusinessException("保存任务进度快照失败");
}
if (isUnchanged(existing, status, totalCount, successCount, failedCount, pendingCount,
currentScopeKey, message, snapshotJson)) {
return;
}
taskProgressSnapshotMapper.update(null, new LambdaUpdateWrapper<TaskProgressSnapshotEntity>()
.eq(TaskProgressSnapshotEntity::getId, existing.getId())
.set(TaskProgressSnapshotEntity::getStatus, status)
@@ -74,6 +79,25 @@ public class TaskProgressSnapshotService {
.set(TaskProgressSnapshotEntity::getUpdatedAt, now));
}
private boolean isUnchanged(TaskProgressSnapshotEntity existing,
String status,
int totalCount,
int successCount,
int failedCount,
int pendingCount,
String currentScopeKey,
String message,
String snapshotJson) {
return Objects.equals(existing.getStatus(), status)
&& Objects.equals(existing.getTotalCount(), totalCount)
&& Objects.equals(existing.getSuccessCount(), successCount)
&& Objects.equals(existing.getFailedCount(), failedCount)
&& Objects.equals(existing.getPendingCount(), pendingCount)
&& Objects.equals(existing.getCurrentScopeKey(), currentScopeKey)
&& Objects.equals(existing.getMessage(), message)
&& Objects.equals(existing.getSnapshotJson(), snapshotJson);
}
public TaskProgressSnapshotEntity find(Long taskId, String moduleType) {
if (taskId == null || taskId <= 0 || isBlank(moduleType)) {
return null;
backend-java/src/main/resources/application.yml
+2
View File
@@ -220,6 +220,8 @@ aiimage:
coze-read-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_READ_TIMEOUT_MILLIS:60000}
coze-poll-interval-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_INTERVAL_MILLIS:30000}
coze-poll-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_TIMEOUT_MILLIS:1800000}
db-task-touch-interval-millis: ${AIIMAGE_SIMILAR_ASIN_DB_TASK_TOUCH_INTERVAL_MILLIS:120000}
db-job-touch-interval-millis: ${AIIMAGE_SIMILAR_ASIN_DB_JOB_TOUCH_INTERVAL_MILLIS:60000}
coze-submit-min-interval-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_SUBMIT_MIN_INTERVAL_MILLIS:5000}
coze-flush-pending-minutes: ${AIIMAGE_SIMILAR_ASIN_COZE_FLUSH_PENDING_MINUTES:1}
coze-submit-max-retry-count: ${AIIMAGE_SIMILAR_ASIN_COZE_SUBMIT_MAX_RETRY_COUNT:5}
backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatServiceTest.java
+26
View File
@@ -4,6 +4,7 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.nanri.aiimage.config.SimilarAsinProperties;
import com.nanri.aiimage.modules.appearancepatent.service.AppearancePatentTaskCacheService;
import com.nanri.aiimage.modules.brand.mapper.BrandCrawlTaskMapper;
import com.nanri.aiimage.modules.brand.service.BrandTaskProgressCacheService;
@@ -31,12 +32,15 @@ import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.time.LocalDateTime;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -62,6 +66,7 @@ class TaskHeartbeatServiceTest {
@Mock private WithdrawTaskCacheService withdrawTaskCacheService;
@Mock private AppearancePatentTaskCacheService appearancePatentTaskCacheService;
@Mock private SimilarAsinTaskCacheService similarAsinTaskCacheService;
@Mock private SimilarAsinProperties similarAsinProperties;
@Mock private DeleteBrandTaskCacheService deleteBrandTaskCacheService;
@Mock private BrandTaskProgressCacheService brandTaskProgressCacheService;
@@ -118,4 +123,25 @@ class TaskHeartbeatServiceTest {
verify(shopDataCrawlTaskCacheService).touchTaskHeartbeat(taskId);
verify(shopDataCrawlTaskCacheService).saveTaskCache(task);
}
@Test
@SuppressWarnings("unchecked")
void similarAsinHeartbeatUsesRedisWithoutRefreshingRecentDatabaseCheckpoint() {
long taskId = 20581L;
FileTaskEntity task = new FileTaskEntity();
task.setId(taskId);
task.setModuleType("SIMILAR_ASIN");
task.setStatus("RUNNING");
task.setUpdatedAt(LocalDateTime.now());
when(fileTaskMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(task);
when(brandCrawlTaskMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(null);
when(similarAsinProperties.getDbTaskTouchIntervalMillis()).thenReturn(120000L);
TaskHeartbeatVo result = service.heartbeat(taskId, new TaskHeartbeatRequest());
assertTrue(result.isAlive());
assertEquals("SIMILAR_ASIN", result.getModuleType());
verify(similarAsinTaskCacheService).touchTaskHeartbeat(taskId);
verify(fileTaskMapper, never()).update(isNull(), any(LambdaUpdateWrapper.class));
}
}
backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotServiceTest.java
+57
View File
@@ -0,0 +1,57 @@
package com.nanri.aiimage.modules.task.service;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.modules.task.mapper.TaskProgressSnapshotMapper;
import com.nanri.aiimage.modules.task.model.entity.TaskProgressSnapshotEntity;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class TaskProgressSnapshotServiceTest {
@BeforeAll
static void initializeMybatisMetadata() {
TableInfoHelper.initTableInfo(
new MapperBuilderAssistant(new MybatisConfiguration(), ""),
TaskProgressSnapshotEntity.class);
}
@Mock private TaskProgressSnapshotMapper taskProgressSnapshotMapper;
@Mock private ObjectMapper objectMapper;
@InjectMocks private TaskProgressSnapshotService service;
@Test
@SuppressWarnings("unchecked")
void saveSkipsDatabaseUpdateWhenSnapshotIsUnchanged() {
TaskProgressSnapshotEntity existing = new TaskProgressSnapshotEntity();
existing.setId(1L);
existing.setStatus("RUNNING");
existing.setTotalCount(10);
existing.setSuccessCount(4);
existing.setFailedCount(1);
existing.setPendingCount(5);
existing.setCurrentScopeKey("scope-1");
existing.setMessage("processing");
when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(existing);
service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 4, 1,
"scope-1", "processing", null);
verify(taskProgressSnapshotMapper, never()).update(isNull(), any(LambdaUpdateWrapper.class));
}
}