Feature/sscs 10722 migration tool performance improvements

build.gradle

@@ -239,7 +239,7 @@ subprojects {
   apply plugin: 'com.jfrog.bintray'
 
   group = 'uk.gov.hmcts.reform.ccd-case-migration'
-  version = '3.0.0'
+  version = 'DEV-SNAPSHOT'
 
   sourceCompatibility = '11'
   targetCompatibility = '11'
@@ -257,6 +257,10 @@ subprojects {
     mavenCentral()
     jcenter()
 
+    // jitpack should be last resort
+    // see: https://github.com/jitpack/jitpack.io/issues/1939
+    maven { url 'https://jitpack.io' }
+
     maven { url "https://dl.bintray.com/hmcts/hmcts-maven" }
     maven { url "https://repo.maven.apache.org/maven2" }
   }

modules/processor/build.gradle

@@ -5,13 +5,14 @@ plugins {
 mainClassName = 'uk.gov.hmcts.reform.migration.CaseMigrationRunner'
 
 dependencies {
-  implementation group: 'org.springframework', name: 'spring-context-support'
-  implementation group: 'com.fasterxml.jackson.module', name: 'jackson-module-parameter-names'
-  implementation group: 'uk.gov.hmcts.reform', name: 'java-logging', version: '5.0.1'
-  implementation group: 'uk.gov.hmcts.reform', name: 'idam-client', version: '1.0.2'
-  implementation group: 'uk.gov.hmcts.reform', name: 'service-auth-provider-client', version: '3.0.0'
-  implementation group: 'uk.gov.hmcts.reform', name: 'core-case-data-store-client', version: '4.7.3'
-  implementation group: 'io.rest-assured', name: 'rest-assured', version: '3.3.0'
+  compile group: 'org.springframework', name: 'spring-context-support'
+  compile group: 'com.fasterxml.jackson.module', name: 'jackson-module-parameter-names'
+  compile group: 'uk.gov.hmcts.reform', name: 'java-logging', version: '5.0.1'
+  compile group: 'uk.gov.hmcts.reform', name: 'idam-client', version: '1.0.2'
+  compile group: 'uk.gov.hmcts.reform', name: 'service-auth-provider-client', version: '3.0.0'
+  compile group: 'com.github.hmcts', name: 'ccd-client', version: '4.8.6'
+  compile group: 'org.elasticsearch', name: 'elasticsearch', version: '7.16.3'
+  compile group: 'io.rest-assured', name: 'rest-assured', version: '3.3.0'
 
   testImplementation group: 'org.springframework.boot', name: 'spring-boot-starter-test'
   testImplementation group: 'com.github.tomakehurst', name: 'wiremock', version: '2.23.2'

modules/processor/src/main/java/uk/gov/hmcts/reform/migration/CaseMigrationProcessor.java

@@ -1,65 +1,190 @@
 package uk.gov.hmcts.reform.migration;
 
+import java.time.LocalDate;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
 import lombok.Getter;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.time.StopWatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 import uk.gov.hmcts.reform.ccd.client.model.CaseDetails;
-import uk.gov.hmcts.reform.migration.service.DataMigrationService;
+import uk.gov.hmcts.reform.ccd.client.model.SearchResult;
 import uk.gov.hmcts.reform.migration.ccd.CoreCaseDataService;
+import uk.gov.hmcts.reform.migration.service.DataMigrationService;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
+import static uk.gov.hmcts.reform.migration.queries.CcdElasticSearchQueries.fetchAllUnsetCaseAccessManagementFieldsCasesQuery;
 
-@Slf4j
 @Component
+@RequiredArgsConstructor
 public class CaseMigrationProcessor {
+
+    private final Logger errorLogger = LoggerFactory.getLogger("ccd-migration-error");
+
+    private final Logger infoLogger = LoggerFactory.getLogger("ccd-migration-info");
+
     private static final String EVENT_ID = "migrateCase";
     private static final String EVENT_SUMMARY = "Migrate Case";
     private static final String EVENT_DESCRIPTION = "Migrate Case";
 
-    @Autowired
-    private CoreCaseDataService coreCaseDataService;
-    @Autowired
-    private DataMigrationService<?> dataMigrationService;
+    private final StopWatch totalTimer = new StopWatch();
+
+    private final CoreCaseDataService coreCaseDataService;
+    private final DataMigrationService<?> dataMigrationService;
+
+    @Getter
+    private final Set<Long> migratedCases = new HashSet<>();
+
     @Getter
-    private List<Long> migratedCases = new ArrayList<>();
+    private final Set<Long> failedCases = new HashSet<>();
+
     @Getter
-    private List<Long> failedCases = new ArrayList<>();
+    private Long totalCases = 0L;
 
-    public void processSingleCase(String userToken, String caseId) {
+    public void processSingleCase(String userToken, String caseId, boolean dryRun) {
         CaseDetails caseDetails;
         try {
             caseDetails = coreCaseDataService.fetchOne(userToken, caseId);
         } catch (Exception ex) {
-            log.error("Case {} not found due to: {}", caseId, ex.getMessage());
+            errorLogger.error("Case {} not found due to: {}", caseId, ex.getMessage());
             return;
         }
         if (dataMigrationService.accepts().test(caseDetails)) {
-            updateCase(userToken, caseDetails.getId(), caseDetails.getData());
+            updateCase(userToken, caseDetails.getId(), caseDetails.getData(), dryRun);
+        } else {
+            infoLogger.info("Case {} already migrated", caseDetails.getId());
+        }
+    }
+
+    public void fetchAndProcessCases(String userToken, boolean dryRun, int numThreads, MigrationPageParams pageParams)
+        throws InterruptedException {
+
+        SearchResult initialSearch = coreCaseDataService.searchCases(userToken,
+            fetchAllUnsetCaseAccessManagementFieldsCasesQuery());
+
+        if (initialSearch.getTotal() <= 0) {
+            return;
+        }
+
+        totalTimer.start();
+
+        int totalCasesToProcess = resolveTotalCasesToProcess(initialSearch, pageParams.getMaxCasesToProcess());
+
+        Long searchFrom = handleFirstCase(userToken, dryRun, initialSearch);
+
+        ExecutorService executorService = Executors.newFixedThreadPool(numThreads);
+
+        fetchAndSubmitTasks(userToken, dryRun, totalCasesToProcess, pageParams.getPageSize(), searchFrom,
+            executorService);
+
+        executorService.shutdown();
+        executorService.awaitTermination(Integer.MAX_VALUE, TimeUnit.DAYS);
+    }
+
+    private int resolveTotalCasesToProcess(SearchResult initialSearch, int maxCasesToProcess) {
+        int totalCasesToProcess = 0;
+
+        if (maxCasesToProcess > 0) {
+            infoLogger.info("Manual case override in use, limiting to {} cases", maxCasesToProcess);
+            totalCasesToProcess = maxCasesToProcess;
         } else {
-            log.info("Case {} already migrated", caseDetails.getId());
+            infoLogger.info("No manual case override in use, fetching all: {} cases", initialSearch.getTotal());
+            totalCasesToProcess = initialSearch.getTotal();
+        }
+
+        return totalCasesToProcess;
+    }
+
+    private void fetchAndSubmitTasks(String userToken, boolean dryRun, int totalCasesToProcess, int pageSize,
+                                     Long searchFrom, ExecutorService executorService) {
+        int casesFetched = 1;
+        int numCasesToFetch = pageSize;
+
+        while (casesFetched < totalCasesToProcess) {
+            numCasesToFetch = resolvePageSize(totalCasesToProcess, casesFetched, numCasesToFetch, pageSize);
+
+            List<CaseDetails> caseDetails =
+                coreCaseDataService.fetchNCases(userToken, numCasesToFetch, searchFrom);
+
+            if (caseDetails.isEmpty()) {
+                break;
+            }
+
+            searchFrom = caseDetails.get(caseDetails.size() - 1).getId();
+
+            executorService.execute(() -> caseDetails
+                .forEach(caseDetail ->
+                    updateCase(userToken, caseDetail.getId(), caseDetail.getData(), dryRun)));
+
+            infoLogger.info("New task submitted");
+
+            casesFetched += caseDetails.size();
+
+            infoLogger.info("{} cases fetched out of {}", casesFetched, totalCasesToProcess);
         }
     }
 
-    public void processAllCases(String userToken, String userId) {
+    private int resolvePageSize(int totalCasesToProcess, int casesFetched, int numCasesToFetch, int pageSize) {
+        int remainingCases = totalCasesToProcess - casesFetched;
+        if (remainingCases < pageSize) {
+            numCasesToFetch = remainingCases;
+        }
+        return numCasesToFetch;
+    }
+
+    private Long handleFirstCase(String userToken, boolean dryRun, SearchResult initialSearch) {
+        infoLogger.info("Processing first case...");
+        CaseDetails firstCase = initialSearch.getCases().get(0);
+        updateCase(userToken, firstCase.getId(), firstCase.getData(), dryRun);
+        return firstCase.getId();
+    }
+
+    public void processAllCases(String userToken, String userId, boolean dryRun) {
         coreCaseDataService.fetchAll(userToken, userId).stream()
             .filter(dataMigrationService.accepts())
-            .forEach(caseDetails -> updateCase(userToken, caseDetails.getId(), caseDetails.getData()));
+            .forEach(caseDetails -> updateCase(userToken, caseDetails.getId(), caseDetails.getData(), dryRun));
     }
 
-    private void updateCase(String authorisation, Long id, Map<String, Object> data) {
-        log.info("Updating case {}", id);
+    protected List<LocalDate> getListOfDates(LocalDate startDate, LocalDate endDate) {
+        return startDate
+            .datesUntil(endDate)
+            .collect(Collectors.toList());
+    }
+
+    private void updateCase(String authorisation, Long id, Map<String, Object> data, boolean dryRun) {
+
+        totalCases++;
+
         try {
-            log.debug("Case data: {}", data);
-            coreCaseDataService.update(authorisation, id.toString(),
-                EVENT_ID, EVENT_SUMMARY, EVENT_DESCRIPTION, dataMigrationService.migrate(data));
-            log.info("Case {} successfully updated", id);
+            var migratedData = dataMigrationService.migrate(data, id);
+            if (!dryRun) {
+                coreCaseDataService.update(
+                    authorisation,
+                    id.toString(),
+                    EVENT_ID,
+                    EVENT_SUMMARY,
+                    EVENT_DESCRIPTION,
+                    migratedData);
+                infoLogger.info("Case {} successfully updated", id);
+            }
             migratedCases.add(id);
+
         } catch (Exception e) {
-            log.error("Case {} update failed due to: {}", id, e.getMessage());
+            errorLogger.error("Case {} update failed due to: {}", id, e.getMessage());
             failedCases.add(id);
         }
+
+        if (totalCases % 1000 == 0) {
+            infoLogger.info("----------{} cases migrated in {} minutes ({} seconds)----------", totalCases,
+                totalTimer.getTime(TimeUnit.MINUTES), totalTimer.getTime(TimeUnit.SECONDS));
+        }
     }
 }

modules/processor/src/main/java/uk/gov/hmcts/reform/migration/CaseMigrationRunner.java

@@ -1,29 +1,37 @@
 package uk.gov.hmcts.reform.migration;
 
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
+import java.util.concurrent.TimeUnit;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.time.StopWatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.context.annotation.PropertySource;
 import uk.gov.hmcts.reform.idam.client.IdamClient;
 
-@Slf4j
 @SpringBootApplication
-@PropertySource("classpath:application.properties")
+@RequiredArgsConstructor
 public class CaseMigrationRunner implements CommandLineRunner {
 
+    private final Logger log = LoggerFactory.getLogger("ccd-migration-info");
     @Value("${migration.idam.username}")
     private String idamUsername;
     @Value("${migration.idam.password}")
     private String idamPassword;
     @Value("${migration.caseId}")
     private String ccdCaseId;
-    @Autowired
-    private IdamClient idamClient;
-    @Autowired
-    private CaseMigrationProcessor caseMigrationProcessor;
+    @Value("${migration.dryrun}")
+    private boolean dryrun;
+    @Value("${migration.pageSize}")
+    private int pageSize;
+    @Value("${migration.maxCasesToProcess}")
+    private int maxCasesToProcess;
+    @Value("${migration.numThreads}")
+    private int numThreads;
+    private final IdamClient idamClient;
+    private final CaseMigrationProcessor caseMigrationProcessor;
 
     public static void main(String[] args) {
         SpringApplication.run(CaseMigrationRunner.class, args);
@@ -33,26 +41,39 @@ public static void main(String[] args) {
     public void run(String... args) {
         try {
             String userToken = idamClient.authenticateUser(idamUsername, idamPassword);
-            log.debug("User token: {}", userToken);
-            String userId = idamClient.getUserDetails(userToken).getId();
-            log.debug("User ID: {}", userId);
+
+            MigrationPageParams pageParams = new MigrationPageParams(pageSize, maxCasesToProcess);
+
+            StopWatch stopWatch = StopWatch.createStarted();
 
             if (ccdCaseId != null && !ccdCaseId.isBlank()) {
                 log.info("Data migration of single case started");
-                caseMigrationProcessor.processSingleCase(userToken, ccdCaseId);
+                caseMigrationProcessor.processSingleCase(userToken, ccdCaseId, dryrun);
             } else {
-                log.info("Data migration of all cases started");
-                caseMigrationProcessor.processAllCases(userToken, userId);
+                log.info("Data migration of cases started");
+                caseMigrationProcessor.fetchAndProcessCases(userToken, dryrun, numThreads, pageParams);
             }
 
+            stopWatch.stop();
+
             log.info("-----------------------------------------");
-            log.info("Data migration completed");
-            log.info("-----------------------------------------");
-            log.info("Total number of processed cases: {}", caseMigrationProcessor.getMigratedCases().size() + caseMigrationProcessor.getFailedCases().size());
+            log.info("Total number of cases: {}", caseMigrationProcessor.getTotalCases());
+            log.info("Total number of processed cases: {}", caseMigrationProcessor.getMigratedCases().size()
+                + caseMigrationProcessor.getFailedCases().size());
             log.info("Total number of migrations performed: {}", caseMigrationProcessor.getMigratedCases().size());
             log.info("-----------------------------------------");
-            log.info("Migrated cases: {} ", !caseMigrationProcessor.getMigratedCases().isEmpty() ? caseMigrationProcessor.getMigratedCases() : "NONE");
-            log.info("Failed cases: {}", !caseMigrationProcessor.getFailedCases().isEmpty() ? caseMigrationProcessor.getFailedCases() : "NONE");
+            log.info("Number of migrated cases: {}", caseMigrationProcessor.getMigratedCases().size());
+            log.info("Migrated cases: {} ", !caseMigrationProcessor.getMigratedCases().isEmpty()
+                ? caseMigrationProcessor.getMigratedCases()
+                : "NONE");
+            log.info("Number of failed cases: {}", caseMigrationProcessor.getFailedCases().size());
+            log.info("Failed cases: {} ", !caseMigrationProcessor.getFailedCases().isEmpty()
+                ? caseMigrationProcessor.getFailedCases()
+                : "NONE");
+            log.info("-----------------------------------------");
+            log.info("Data migration completed in: {} minutes ({} seconds).",
+                stopWatch.getTime(TimeUnit.MINUTES), stopWatch.getTime(TimeUnit.SECONDS));
+            log.info("-----------------------------------------");
         } catch (Throwable e) {
             log.error("Migration failed with the following reason: {}", e.getMessage(), e);
         }

modules/processor/src/main/java/uk/gov/hmcts/reform/migration/MigrationPageParams.java

@@ -0,0 +1,14 @@
+package uk.gov.hmcts.reform.migration;
+
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+@Getter
+@RequiredArgsConstructor
+public class MigrationPageParams {
+
+    private final int pageSize;
+
+    private final int maxCasesToProcess;
+}