diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java index 0829ec8..1718727 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java @@ -8,7 +8,7 @@ import ru.nbch.credit_tracker.entities.app.PhonesMap; @Mapper public interface PhonesMapMapper { List getCachedData(@Param("phones") List phones); - List searchedAtLessThan(@Param("daysCount") Integer daysCount, @Param("limit") int limit, @Param("offset") int offset); + List searchedAtLessThan(@Param("daysCount") Integer daysCount, @Param("batchSize") int batchSize, @Param("lastId") Long lastId); PhonesMap findByPhone(@Param("phone") Long phone); List findEligibleForFlagsUpdate(@Param("lastId") Long lastId, @Param("batchSize") Integer batchSize); @@ -16,6 +16,7 @@ public interface PhonesMapMapper { void updateNameCachedData(@Param("json") String json); void updateFlagCachedData(@Param("json") String json); void resetFidAndPD(@Param("json") String json); + void updateSearchedAt(@Param("json") String json); void registerJson(@Param("json") String json); void setFidJson(@Param("json") String json); void deleteUnusedPhones(); diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java index 7ee1d9a..17beb63 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java @@ -94,17 +94,4 @@ public class UpdateProcessingPackagesService { } } - private void processingRepeatedRestore() { - List packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING_REPEATED.name()); - log.debug("Found {} packages in status PROCESSING_REPEATED", packages.size()); - for (UserPackage userPackage : packages) { - try { - Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.REPEATED_RESTORE)); - pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId()); - } catch (Exception e) { - log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e)); - } - } - } - } diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java index 47aae57..37b7735 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java @@ -42,7 +42,11 @@ public class UpdatingFidAndPdService { @Scheduled(cron = "${ct.update-fid-cron-expression}") public void update() { - log.info("Delete delete unused phones"); + if (!pipelineRecalcPDOrchestrator.checkPermits()) { + log.info("The previous task has not yet been completed."); + return; + } + log.info("Delete unused phones"); phonesMapService.deleteUnusedPhones(); log.info("Start updating fids and PD"); try { diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java index 065784e..8f80fa7 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java @@ -33,6 +33,19 @@ public class PhonesMapService { } } + public void updateSearchedAt(List phones) { + try { + String json = entityToJsonMapper.writeValueAsString(phones.stream() + .filter(Objects::nonNull) + .map(phone -> new HashMap() {{ + put("phone", phone); + }}).toList()); + phonesMapMapper.updateSearchedAt(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + public void updateNameCachedData(List subjectProfiles) { try { String json = entityToJsonMapper.writeValueAsString(subjectProfiles.stream() diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java index f024ed6..a2964ef 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java @@ -36,6 +36,11 @@ public class PipelineRecalcPDOrchestrator { return innerStart(pipeline); } + public boolean checkPermits() { + synchronized (lock) { + return semaphore.availablePermits() > 0; + } + } private CompletableFuture innerStart(Pipeline pipeline) { try { pipeline.start(pipelineExecutorService); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java index 5123d05..b952151 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java @@ -51,10 +51,12 @@ public class FidIdEnricher implements Handler { log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size()); List noIdDataFids = subjectsProfile.stream() - .filter(subjectProfile -> subjectProfile.getIdData() == null) + .filter(subjectProfile -> subjectProfile.getIdData() == null && subjectProfile.getFid() != null) .map(SubjectProfile::getFid) .toList(); + long countEmptyFid = subjectsProfile.stream().filter(subjectProfile -> subjectProfile.getFid() == null).count(); + log.info("Count with empty fid: {}", countEmptyFid); Map idByFid; if (!noIdDataFids.isEmpty()) { idByFid = personService.filledPassportsData(noIdDataFids); @@ -64,6 +66,7 @@ public class FidIdEnricher implements Handler { List idCacheOnUpdate = new ArrayList<>(); List personOnProcessing = subjectsProfile.stream() + .filter(subjectProfile -> subjectProfile.getFid() != null) .flatMap(subjectProfile -> { if (subjectProfile.getIdData() == null) { Id id = idByFid.get(subjectProfile.getFid()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java index 896eda6..bce267a 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java @@ -1,6 +1,5 @@ package ru.nbch.credit_tracker.service.task.impl; -import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.stream.Stream; @@ -36,14 +35,13 @@ public class FidIdUpdater implements Handler { @Override public VoidPayload handle(SubjectPayload taskPayload) { List subjectsProfile = taskPayload.getSubjects(); - log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size()); + log.debug("FidIdUpdater handle; size: {}", subjectsProfile.size()); List noIdDataFids = subjectsProfile.stream() .map(SubjectProfile::getFid) .toList(); Map idByFid = personService.filledPassportsData(noIdDataFids); - List idCacheOnUpdate = new ArrayList<>(); List personOnProcessing = subjectsProfile.stream() .flatMap(subjectProfile -> { Id id = idByFid.get(subjectProfile.getFid()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java index cb2f859..1e44469 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java @@ -3,6 +3,7 @@ package ru.nbch.credit_tracker.service.task.impl; import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.stream.Stream; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; @@ -47,7 +48,7 @@ public class FidNameUpdater implements Handler { List changingFidPhones = new ArrayList<>(); List notFoundFidPhones = new ArrayList<>(); - subjects.forEach(subjectProfile -> { + List notChangedPhones = subjects.stream().flatMap(subjectProfile -> { Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone())); if (name != null) { if (subjectProfile.getFid() == null || !subjectProfile.getFid().equals(name.getFid())) { @@ -57,11 +58,14 @@ public class FidNameUpdater implements Handler { subjectProfile.getNameData().setMiddleName(name.getMiddle()); subjectProfile.getNameData().setBirthDate(name.getBirthDate()); changingFidPhones.add(subjectProfile); + return Stream.empty(); } } else if (subjectProfile.getFid() != null) { notFoundFidPhones.add(subjectProfile.getPhone()); + return Stream.empty(); } - }); + return Stream.of(subjectProfile.getPhone()); + }).toList(); List changingFids = new ArrayList<>(); if (!notFoundFidPhones.isEmpty()) { phonesMapService.resetFidAndPDByPhones(notFoundFidPhones); @@ -73,6 +77,10 @@ public class FidNameUpdater implements Handler { changingFids.addAll(changingFidPhones.stream().map(SubjectProfile::getPhone).toList()); } + if (!notChangedPhones.isEmpty()) { + phonesMapService.updateSearchedAt(notChangedPhones); + } + if (!changingFids.isEmpty()) { List packageIds = packageRecordsService.selectUniquePackageIds(changingFids); if (!packageIds.isEmpty()) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java index f066860..e07f862 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java @@ -64,8 +64,8 @@ public class KickRecalcPipeLine implements IPipeLineStarter { public void startFillingQueue(BlockingQueue firstQueue) { try { log.info("Start kicker: mode:{};", mode()); - int offset = 0; - List page = nextPage(offset); + long lastId = 0; + List page = nextPage(lastId); while (!page.isEmpty()) { log.debug("Get page {}", page.size()); List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() @@ -92,8 +92,8 @@ public class KickRecalcPipeLine implements IPipeLineStarter { log.error("Pipeline starter was interrupted!"); throw new PipelineStarterException(); } - offset += BATCH_SIZE; - page = nextPage(offset); + lastId = page.getLast().getId(); + page = nextPage(lastId); } } catch (PipelineStarterException e) { log.info("Pipeline starter shutting down..."); @@ -115,7 +115,7 @@ public class KickRecalcPipeLine implements IPipeLineStarter { protected PipelineMode mode() { return PipelineMode.REPEATED_RESTORE; } - protected List nextPage(int offset) { - return phonesMapMapper.searchedAtLessThan(daysToUpdateFids.intValue(), BATCH_SIZE, offset); + protected List nextPage(long lastId) { + return phonesMapMapper.searchedAtLessThan(daysToUpdateFids.intValue(), BATCH_SIZE, lastId); } } diff --git a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml index 39a84f7..b9c409f 100644 --- a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml @@ -54,24 +54,24 @@ searched_at, calculated_at FROM signals_phones_map - WHERE searched_at < now() - make_interval(days => #{daysCount}) - ORDER BY id - LIMIT #{limit} OFFSET #{offset} + WHERE searched_at < now() - make_interval(days => #{daysCount}) and id > #{lastId} + order by id + limit #{batchSize}