diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/service/NodeSharedStore.java b/backend-java/src/main/java/com/nanri/aiimage/common/service/NodeSharedStore.java index 5a00d370..72f208b4 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/common/service/NodeSharedStore.java +++ b/backend-java/src/main/java/com/nanri/aiimage/common/service/NodeSharedStore.java @@ -121,6 +121,16 @@ public final class NodeSharedStore { return local.size(); } + /** + * 本节点已知条目的快照（副本，可安全遍历）。 + * + *

用途：保留期清理等维护动作需要遍历键；跨节点的过期回收由 Redis TTL 兜底， + * 因此这里只返回本节点写入过的条目即可。 + */ + public java.util.Map localEntriesSnapshot() { + return new java.util.LinkedHashMap<>(local); + } + private boolean throttleAllowsWrite(K key) { if (writeThrottleMillis <= 0L) { return true; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java index 27cddfed..6e64a1c6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java @@ -2,7 +2,9 @@ package com.nanri.aiimage.modules.dedupe.service; import cn.hutool.core.util.IdUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.service.NodeSharedStore; import com.nanri.aiimage.common.util.ExcelStreamReader; import com.nanri.aiimage.modules.dedupe.mapper.DedupeTotalDataMapper; import com.nanri.aiimage.modules.dedupe.model.dto.DedupeTotalDataCreateRequest; @@ -63,7 +65,6 @@ import java.util.zip.ZipEntry; import java.util.zip.ZipOutputStream; @Service -@RequiredArgsConstructor @Slf4j public class DedupeTotalDataService { @@ -73,18 +74,60 @@ public class DedupeTotalDataService { private static final long COMPLETED_PROGRESS_RETENTION_MILLIS = 60 * 60 * 1000L; private static final DateTimeFormatter EXPORT_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + /** 导入进度等状态的 TTL：完成后仍需可查一段时间（与原内存缓存 1 小时保留对齐并留余量）。 */ + private static final java.time.Duration IMPORT_STATE_TTL = java.time.Duration.ofHours(3); + private final DedupeTotalDataMapper dedupeTotalDataMapper; private final AdminUserMapper adminUserMapper; private final ShopManageGroupMapper shopManageGroupMapper; private final PlatformTransactionManager transactionManager; - private final Map importProgressMap = new ConcurrentHashMap<>(); - private final Map deleteImportProgressMap = new ConcurrentHashMap<>(); - private final Map importOwnerMap = new ConcurrentHashMap<>(); - private final Map deleteImportOwnerMap = new ConcurrentHashMap<>(); - private final Map importGroupMap = new ConcurrentHashMap<>(); - private final Map deleteImportGroupMap = new ConcurrentHashMap<>(); - private final Map importCompletedAtMap = new ConcurrentHashMap<>(); - private final Map deleteImportCompletedAtMap = new ConcurrentHashMap<>(); + + // 2026-09 全维度审查 D2：进度/归属/分组改由 NodeSharedStore 承载（本地快路径 + Redis 跨节点）， + // 此前只存节点本地内存，轮询落到另一节点会报"任务不存在"（nginx 的 user_id 亲和只覆盖常态）。 + private final NodeSharedStore importProgressMap; + private final NodeSharedStore deleteImportProgressMap; + private final NodeSharedStore importOwnerMap; + private final NodeSharedStore deleteImportOwnerMap; + private final NodeSharedStore importGroupMap; + private final NodeSharedStore deleteImportGroupMap; + private final NodeSharedStore importCompletedAtMap; + private final NodeSharedStore deleteImportCompletedAtMap; + + public DedupeTotalDataService(DedupeTotalDataMapper dedupeTotalDataMapper, + AdminUserMapper adminUserMapper, + ShopManageGroupMapper shopManageGroupMapper, + PlatformTransactionManager transactionManager, + org.springframework.data.redis.core.StringRedisTemplate stringRedisTemplate, + ObjectMapper objectMapper) { + this.dedupeTotalDataMapper = dedupeTotalDataMapper; + this.adminUserMapper = adminUserMapper; + this.shopManageGroupMapper = shopManageGroupMapper; + this.transactionManager = transactionManager; + this.importProgressMap = progressStore("aiimage:dedupe:import:progress", stringRedisTemplate, objectMapper); + this.deleteImportProgressMap = progressStore("aiimage:dedupe:delete-import:progress", stringRedisTemplate, objectMapper); + this.importOwnerMap = longStore("aiimage:dedupe:import:owner", stringRedisTemplate, objectMapper); + this.deleteImportOwnerMap = longStore("aiimage:dedupe:delete-import:owner", stringRedisTemplate, objectMapper); + this.importGroupMap = longStore("aiimage:dedupe:import:group", stringRedisTemplate, objectMapper); + this.deleteImportGroupMap = longStore("aiimage:dedupe:delete-import:group", stringRedisTemplate, objectMapper); + // 完成时间只用于本节点保留期清理，不参与跨节点读取 → 不节流 + this.importCompletedAtMap = longStoreNoThrottle("aiimage:dedupe:import:completed-at", stringRedisTemplate, objectMapper); + this.deleteImportCompletedAtMap = longStoreNoThrottle("aiimage:dedupe:delete-import:completed-at", stringRedisTemplate, objectMapper); + } + + private static NodeSharedStore progressStore( + String prefix, org.springframework.data.redis.core.StringRedisTemplate redis, ObjectMapper mapper) { + return new NodeSharedStore<>(prefix, IMPORT_STATE_TTL, DedupeTotalDataImportProgressVo.class, redis, mapper); + } + + private static NodeSharedStore longStore( + String prefix, org.springframework.data.redis.core.StringRedisTemplate redis, ObjectMapper mapper) { + return new NodeSharedStore<>(prefix, IMPORT_STATE_TTL, Long.class, redis, mapper); + } + + private static NodeSharedStore longStoreNoThrottle( + String prefix, org.springframework.data.redis.core.StringRedisTemplate redis, ObjectMapper mapper) { + return new NodeSharedStore<>(prefix, IMPORT_STATE_TTL, Long.class, redis, mapper, 0L); + } /** 导入任务并发上限，避免 POI 解析和批量数据库写入叠加。 */ private final Semaphore importSlots = new Semaphore(4); @@ -1235,19 +1278,26 @@ public class DedupeTotalDataService { deleteImportGroupMap, cutoff); } + /** + * 清理已结束且超过保留期的进度条目。 + * 只遍历本节点写入过的条目（跨节点的过期由 Redis TTL 兜底），避免为清理做全量 SCAN。 + */ private void cleanupExpiredProgressEntries( - Map completedAtMap, - Map progressMap, - Map ownerMap, - Map groupMap, + NodeSharedStore completedAtStore, + NodeSharedStore progressStore, + NodeSharedStore ownerStore, + NodeSharedStore groupStore, long cutoff) { - completedAtMap.forEach((id, completedAt) -> { - if (completedAt != null && completedAt < cutoff && completedAtMap.remove(id, completedAt)) { - progressMap.remove(id); - ownerMap.remove(id); - groupMap.remove(id); + for (Map.Entry entry : completedAtStore.localEntriesSnapshot().entrySet()) { + Long completedAt = entry.getValue(); + if (completedAt == null || completedAt >= cutoff) { + continue; } - }); + completedAtStore.remove(entry.getKey()); + progressStore.remove(entry.getKey()); + ownerStore.remove(entry.getKey()); + groupStore.remove(entry.getKey()); + } } private String normalizeExcelText(String value) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/QueryAsinService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/QueryAsinService.java index 2dacac38..42e2c1d8 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/QueryAsinService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/QueryAsinService.java @@ -3,7 +3,9 @@ package com.nanri.aiimage.modules.shopkey.service; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.toolkit.support.SFunction; import cn.hutool.core.util.IdUtil; +import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.service.NodeSharedStore; import com.nanri.aiimage.common.util.ExcelStreamReader; import com.nanri.aiimage.config.TaskPressureProperties; import com.nanri.aiimage.modules.shopkey.mapper.QueryAsinMapper; @@ -44,7 +46,6 @@ import java.util.concurrent.Semaphore; import java.util.function.Predicate; @Service -@RequiredArgsConstructor @Slf4j public class QueryAsinService { @@ -60,18 +61,39 @@ public class QueryAsinService { private final QueryAsinMapper queryAsinMapper; private final ShopManageGroupService shopManageGroupService; private final TaskPressureProperties taskPressureProperties; - private final Map importProgressMap = new ConcurrentHashMap<>(); - private final Map deleteImportProgressMap = new ConcurrentHashMap<>(); + // 2026-09 全维度审查 D2：进度/完成时间改由 NodeSharedStore 承载（本地快路径 + Redis 跨节点）， + // 此前只存节点本地内存，轮询落到另一节点会报"任务不存在"。 + private final NodeSharedStore importProgressMap; + private final NodeSharedStore deleteImportProgressMap; + /** 已结束导入的完成时间，用于**本节点**的保留期回收；跨节点过期由 Redis TTL 兜底。 */ + private final NodeSharedStore completedImportAtMap; @Value("${aiimage.shop-key.max-import-file-bytes:104857600}") private long maxImportFileBytes = 100L * 1024 * 1024; @Value("${aiimage.shop-key.max-import-rows:500000}") private int maxImportRows = 500_000; - /** 已结束导入的完成时间，用于定时回收进度快照，避免进程内 Map 无界增长。 */ - private final Map completedImportAtMap = new ConcurrentHashMap<>(); /** 导入任务并发上限，避免多个 POI/数据库导入同时拖垮资源。 */ private final Semaphore importSlots = new Semaphore(4); + public QueryAsinService( QueryAsinMapper queryAsinMapper, + ShopManageGroupService shopManageGroupService, + TaskPressureProperties taskPressureProperties, + org.springframework.data.redis.core.StringRedisTemplate stringRedisTemplate, + ObjectMapper objectMapper) { + this.queryAsinMapper = queryAsinMapper; + this.shopManageGroupService = shopManageGroupService; + this.taskPressureProperties = taskPressureProperties; + this.importProgressMap = new NodeSharedStore<>("aiimage:query-asin:import:progress", + IMPORT_STATE_TTL, QueryAsinImportProgressVo.class, stringRedisTemplate, objectMapper); + this.deleteImportProgressMap = new NodeSharedStore<>("aiimage:query-asin:delete-import:progress", + IMPORT_STATE_TTL, QueryAsinImportProgressVo.class, stringRedisTemplate, objectMapper); + this.completedImportAtMap = new NodeSharedStore<>("aiimage:query-asin:import:completed-at", + IMPORT_STATE_TTL, Long.class, stringRedisTemplate, objectMapper, 0L); + } + + /** 导入状态在 Redis 的保留时长：完成后仍需可查一段时间（原内存保留 1 小时 + 余量）。 */ + private static final java.time.Duration IMPORT_STATE_TTL = java.time.Duration.ofHours(3); + public QueryAsinPageVo page(long page, long pageSize, Long groupId, String shopName, String asin, String country, Long operatorId, boolean superAdmin) { long safePage = Math.max(page, 1); @@ -332,7 +354,8 @@ public class QueryAsinService { } String importId = IdUtil.fastSimpleUUID(); QueryAsinImportProgressVo progress = newImportProgress(); - Map progressMap = deleteMode ? deleteImportProgressMap : importProgressMap; + NodeSharedStore progressMap = + deleteMode ? deleteImportProgressMap : importProgressMap; progressMap.put(importId, progress); try { File tempFile = saveMultipartToTempFile(file, deleteMode ? "query-asin-delete-" : "query-asin-import-"); @@ -364,18 +387,22 @@ public class QueryAsinService { @Scheduled(fixedDelayString = "${aiimage.shop-key.import-progress-cleanup-delay-ms:300000}") public void cleanupCompletedImports() { long cutoff = System.currentTimeMillis() - COMPLETED_IMPORT_RETENTION_MILLIS; - completedImportAtMap.forEach((importId, completedAt) -> { - if (completedAt != null && completedAt < cutoff - && completedImportAtMap.remove(importId, completedAt)) { - importProgressMap.remove(importId); - deleteImportProgressMap.remove(importId); + // 只遍历本节点写入过的条目（跨节点过期由 Redis TTL 兜底），避免为清理做全量 SCAN + for (Map.Entry entry : completedImportAtMap.localEntriesSnapshot().entrySet()) { + Long completedAt = entry.getValue(); + if (completedAt == null || completedAt >= cutoff) { + continue; } - }); + completedImportAtMap.remove(entry.getKey()); + importProgressMap.remove(entry.getKey()); + deleteImportProgressMap.remove(entry.getKey()); + } } private void runImportTask(String importId, File tempFile, String filename, Long fallbackGroupId, Long operatorId, boolean superAdmin, boolean deleteMode) { - Map progressMap = deleteMode ? deleteImportProgressMap : importProgressMap; + NodeSharedStore progressMap = + deleteMode ? deleteImportProgressMap : importProgressMap; QueryAsinImportProgressVo progress = progressMap.get(importId); if (progress == null) { deleteQuietly(tempFile); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/SkipPriceAsinService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/SkipPriceAsinService.java index bfea9b7e..2d96b43c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/SkipPriceAsinService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/SkipPriceAsinService.java @@ -4,7 +4,9 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.toolkit.support.SFunction; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import cn.hutool.core.util.IdUtil; +import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.service.NodeSharedStore; import com.nanri.aiimage.common.util.ExcelStreamReader; import com.nanri.aiimage.modules.shopkey.mapper.SkipPriceAsinMapper; import com.nanri.aiimage.modules.shopkey.mapper.ShopManageMapper; @@ -46,7 +48,6 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Semaphore; @Service -@RequiredArgsConstructor @Slf4j public class SkipPriceAsinService { @@ -60,17 +61,38 @@ public class SkipPriceAsinService { private final SkipPriceAsinMapper skipPriceAsinMapper; private final ShopManageMapper shopManageMapper; private final ShopManageGroupService shopManageGroupService; - private final Map importProgressMap = new ConcurrentHashMap<>(); - private final Map deleteImportProgressMap = new ConcurrentHashMap<>(); + // 2026-09 全维度审查 D2：进度/完成时间改由 NodeSharedStore 承载（本地快路径 + Redis 跨节点）， + // 此前只存节点本地内存，轮询落到另一节点会报"任务不存在"。 + private final NodeSharedStore importProgressMap; + private final NodeSharedStore deleteImportProgressMap; + /** 已结束导入的完成时间，用于**本节点**的保留期回收；跨节点过期由 Redis TTL 兜底。 */ + private final NodeSharedStore completedImportAtMap; @Value("${aiimage.shop-key.max-import-file-bytes:104857600}") private long maxImportFileBytes = 100L * 1024 * 1024; @Value("${aiimage.shop-key.max-import-rows:500000}") private int maxImportRows = 500_000; - /** 已结束导入的完成时间，用于定时回收进度快照，避免进程内 Map 无界增长。 */ - private final Map completedImportAtMap = new ConcurrentHashMap<>(); /** 导入任务并发上限，避免多个 POI/数据库导入同时拖垮资源。 */ private final Semaphore importSlots = new Semaphore(4); + + public SkipPriceAsinService( SkipPriceAsinMapper skipPriceAsinMapper, + ShopManageMapper shopManageMapper, + ShopManageGroupService shopManageGroupService, + org.springframework.data.redis.core.StringRedisTemplate stringRedisTemplate, + ObjectMapper objectMapper) { + this.skipPriceAsinMapper = skipPriceAsinMapper; + this.shopManageMapper = shopManageMapper; + this.shopManageGroupService = shopManageGroupService; + this.importProgressMap = new NodeSharedStore<>("aiimage:skip-price-asin:import:progress", + IMPORT_STATE_TTL, QueryAsinImportProgressVo.class, stringRedisTemplate, objectMapper); + this.deleteImportProgressMap = new NodeSharedStore<>("aiimage:skip-price-asin:delete-import:progress", + IMPORT_STATE_TTL, QueryAsinImportProgressVo.class, stringRedisTemplate, objectMapper); + this.completedImportAtMap = new NodeSharedStore<>("aiimage:skip-price-asin:import:completed-at", + IMPORT_STATE_TTL, Long.class, stringRedisTemplate, objectMapper, 0L); + } + + /** 导入状态在 Redis 的保留时长：完成后仍需可查一段时间（原内存保留 1 小时 + 余量）。 */ + private static final java.time.Duration IMPORT_STATE_TTL = java.time.Duration.ofHours(3); private final Map skipAsinLookupCache = new ConcurrentHashMap<>(); public SkipPriceAsinPageVo page(long page, long pageSize, Long groupId, String shopName, String asin, @@ -410,7 +432,8 @@ public class SkipPriceAsinService { } String importId = IdUtil.fastSimpleUUID(); QueryAsinImportProgressVo progress = newImportProgress(); - Map progressMap = deleteMode ? deleteImportProgressMap : importProgressMap; + NodeSharedStore progressMap = + deleteMode ? deleteImportProgressMap : importProgressMap; progressMap.put(importId, progress); try { File tempFile = saveMultipartToTempFile(file, deleteMode ? "skip-price-asin-delete-" : "skip-price-asin-import-"); @@ -442,18 +465,22 @@ public class SkipPriceAsinService { @Scheduled(fixedDelayString = "${aiimage.shop-key.import-progress-cleanup-delay-ms:300000}") public void cleanupCompletedImports() { long cutoff = System.currentTimeMillis() - COMPLETED_IMPORT_RETENTION_MILLIS; - completedImportAtMap.forEach((importId, completedAt) -> { - if (completedAt != null && completedAt < cutoff - && completedImportAtMap.remove(importId, completedAt)) { - importProgressMap.remove(importId); - deleteImportProgressMap.remove(importId); + // 只遍历本节点写入过的条目（跨节点过期由 Redis TTL 兜底），避免为清理做全量 SCAN + for (Map.Entry entry : completedImportAtMap.localEntriesSnapshot().entrySet()) { + Long completedAt = entry.getValue(); + if (completedAt == null || completedAt >= cutoff) { + continue; } - }); + completedImportAtMap.remove(entry.getKey()); + importProgressMap.remove(entry.getKey()); + deleteImportProgressMap.remove(entry.getKey()); + } } private void runImportTask(String importId, File tempFile, String filename, Long groupId, String shopName, boolean deleteMode) { - Map progressMap = deleteMode ? deleteImportProgressMap : importProgressMap; + NodeSharedStore progressMap = + deleteMode ? deleteImportProgressMap : importProgressMap; QueryAsinImportProgressVo progress = progressMap.get(importId); if (progress == null) { deleteQuietly(tempFile); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataServiceTest.java index 56c77e5e..80afcdee 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataServiceTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataServiceTest.java @@ -7,6 +7,7 @@ import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.modules.dedupe.mapper.DedupeTotalDataMapper; import com.nanri.aiimage.modules.dedupe.model.dto.DedupeTotalDataCreateRequest; import com.nanri.aiimage.modules.dedupe.model.entity.DedupeTotalDataEntity; +import com.nanri.aiimage.common.service.NodeSharedStore; import com.nanri.aiimage.modules.dedupe.model.vo.DedupeTotalDataImportProgressVo; import com.nanri.aiimage.modules.dedupe.model.vo.DedupeTotalDataItemVo; import com.nanri.aiimage.modules.dedupe.model.vo.DedupeTotalDataPageVo; @@ -481,38 +482,45 @@ class DedupeTotalDataServiceTest { @Test @SuppressWarnings("unchecked") void expiredCompletedProgressIsRemovedOnNextLookup() { - Map progressMap = - (Map) ReflectionTestUtils.getField(service, "importProgressMap"); - Map ownerMap = (Map) ReflectionTestUtils.getField(service, "importOwnerMap"); - Map groupMap = (Map) ReflectionTestUtils.getField(service, "importGroupMap"); - Map completedAtMap = - (Map) ReflectionTestUtils.getField(service, "importCompletedAtMap"); + NodeSharedStore progressStore = + (NodeSharedStore) ReflectionTestUtils.getField(service, "importProgressMap"); + NodeSharedStore ownerStore = + (NodeSharedStore) ReflectionTestUtils.getField(service, "importOwnerMap"); + NodeSharedStore groupStore = + (NodeSharedStore) ReflectionTestUtils.getField(service, "importGroupMap"); + NodeSharedStore completedAtStore = + (NodeSharedStore) ReflectionTestUtils.getField(service, "importCompletedAtMap"); DedupeTotalDataImportProgressVo progress = new DedupeTotalDataImportProgressVo(); progress.setStatus("success"); - progressMap.put("expired", progress); - ownerMap.put("expired", 23L); - groupMap.put("expired", 7L); - completedAtMap.put("expired", System.currentTimeMillis() - (2 * 60 * 60 * 1000L)); + progressStore.put("expired", progress); + ownerStore.put("expired", 23L); + groupStore.put("expired", 7L); + completedAtStore.put("expired", System.currentTimeMillis() - (2 * 60 * 60 * 1000L)); assertThrows(BusinessException.class, () -> service.getImportProgress("expired", 23L)); - assertFalse(progressMap.containsKey("expired")); - assertFalse(ownerMap.containsKey("expired")); - assertFalse(groupMap.containsKey("expired")); - assertFalse(completedAtMap.containsKey("expired")); + assertFalse(progressStore.containsKey("expired")); + assertFalse(ownerStore.containsKey("expired")); + assertFalse(groupStore.containsKey("expired")); + assertFalse(completedAtStore.containsKey("expired")); } + /** + * 直接注入导入进度状态（D2 后由 NodeSharedStore 承载：测试里未注入 Redis，等价于纯本地）。 + */ @SuppressWarnings("unchecked") private DedupeTotalDataImportProgressVo registerImportProgress(String importId, Long ownerId, Long groupId) { - Map progressMap = - (Map) ReflectionTestUtils.getField(service, "importProgressMap"); - Map ownerMap = (Map) ReflectionTestUtils.getField(service, "importOwnerMap"); - Map groupMap = (Map) ReflectionTestUtils.getField(service, "importGroupMap"); + NodeSharedStore progressStore = + (NodeSharedStore) ReflectionTestUtils.getField(service, "importProgressMap"); + NodeSharedStore ownerStore = + (NodeSharedStore) ReflectionTestUtils.getField(service, "importOwnerMap"); + NodeSharedStore groupStore = + (NodeSharedStore) ReflectionTestUtils.getField(service, "importGroupMap"); DedupeTotalDataImportProgressVo progress = new DedupeTotalDataImportProgressVo(); progress.setStatus("running"); - progressMap.put(importId, progress); - ownerMap.put(importId, ownerId); - groupMap.put(importId, groupId); + progressStore.put(importId, progress); + ownerStore.put(importId, ownerId); + groupStore.put(importId, groupId); return progress; }