From 62fa7ec78704a3b7ec58aaa24454a796d8524a53 Mon Sep 17 00:00:00 2001 From: TipzCM Date: Tue, 1 Oct 2024 13:52:39 -0400 Subject: [PATCH] value set expansion (#6286) --- .../src/checkstyle/hapi-base-checkstyle.xml | 8 + .../6285-batch2-reindex-version2-added.yaml | 17 ++ .../fhir/jpa/dao/BaseHapiFhirResourceDao.java | 4 +- .../jpa/dao/JpaResourceDaoCodeSystem.java | 45 ++++ .../term/TermCodeSystemStorageSvcImpl.java | 2 +- .../jpa/term/TermDeferredStorageSvcImpl.java | 18 +- .../jpa/term/api/ITermDeferredStorageSvc.java | 2 + ...rceProviderCustomSearchParamDstu3Test.java | 11 +- ...FhirResourceDaoR4ComboUniqueParamTest.java | 3 +- ...esourceDaoR4IndexStorageOptimizedTest.java | 7 +- .../r4/FhirResourceDaoR4QueryCountTest.java | 32 ++- ...ourceDaoR4SearchCustomSearchParamTest.java | 7 +- ...exStepTest.java => ReindexStepV1Test.java} | 8 +- ...sourceProviderCustomSearchParamR4Test.java | 9 +- ...exStepTest.java => ReindexStepV1Test.java} | 81 +++++-- ...indexJobTest.java => ReindexTaskTest.java} | 34 +-- ...a => ReindexTaskWithPartitioningTest.java} | 8 +- .../jpa/term/ValueSetExpansionR4Test.java | 180 +++++++++++++++- .../fhir/jpa/dao/r5/DuplicateIndexR5Test.java | 4 +- .../provider/r5/ResourceProviderR5Test.java | 4 +- .../ca/uhn/fhir/jpa/test/BaseJpaR4Test.java | 3 + .../jpa/test/PatientReindexTestHelper.java | 3 +- .../term/TermDeferredStorageSvcImplTest.java | 15 +- .../batch2/jobs/reindex/ReindexAppCtx.java | 77 +++---- .../batch2/jobs/reindex/ReindexProvider.java | 3 +- .../fhir/batch2/jobs/reindex/ReindexStep.java | 203 ------------------ .../batch2/jobs/reindex/ReindexUtils.java | 39 ++++ .../jobs/reindex/models/ReindexResults.java | 29 +++ .../jobs/reindex/svcs/ReindexJobService.java | 38 ++++ .../v1/ReindexJobParametersValidatorV1.java | 57 +++++ .../batch2/jobs/reindex/v1/ReindexStepV1.java | 117 ++++++++++ .../batch2/jobs/reindex/v1/ReindexTaskV1.java | 202 +++++++++++++++++ .../jobs/reindex/v1/ReindexV1Config.java | 101 +++++++++ .../v2/CheckPendingReindexWorkStep.java | 42 ++++ .../ReindexJobParametersValidatorV2.java} | 7 +- .../batch2/jobs/reindex/v2/ReindexStepV2.java | 118 ++++++++++ .../batch2/jobs/reindex/v2/ReindexTaskV2.java | 201 +++++++++++++++++ .../jobs/reindex/v2/ReindexV2Config.java | 110 ++++++++++ ...> ReindexTaskParametersValidatorTest.java} | 5 +- .../batch2/api/RetryChunkLaterException.java | 11 +- .../chunk/ResourceIdListWorkChunkJson.java | 2 +- .../fhir/jpa/api/dao/IFhirResourceDao.java | 13 ++ .../uhn/fhir/jpa/api/dao/ReindexOutcome.java | 13 ++ .../fhir/jpa/api/model/ReindexJobStatus.java | 16 ++ 44 files changed, 1550 insertions(+), 359 deletions(-) create mode 100644 hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/7_6_0/6285-batch2-reindex-version2-added.yaml rename hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/tx/{ReindexStepTest.java => ReindexStepV1Test.java} (92%) rename hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/{ReindexStepTest.java => ReindexStepV1Test.java} (80%) rename hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/{ReindexJobTest.java => ReindexTaskTest.java} (95%) rename hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/{ReindexJobWithPartitioningTest.java => ReindexTaskWithPartitioningTest.java} (97%) delete mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexStep.java create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexUtils.java create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/models/ReindexResults.java create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/svcs/ReindexJobService.java create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexJobParametersValidatorV1.java create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexStepV1.java create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexTaskV1.java create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexV1Config.java create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/CheckPendingReindexWorkStep.java rename hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/{ReindexJobParametersValidator.java => v2/ReindexJobParametersValidatorV2.java} (85%) create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexStepV2.java create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexTaskV2.java create mode 100644 hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexV2Config.java rename hapi-fhir-storage-batch2-jobs/src/test/java/ca/uhn/fhir/batch2/jobs/reindex/{ReindexJobParametersValidatorTest.java => ReindexTaskParametersValidatorTest.java} (86%) create mode 100644 hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/model/ReindexJobStatus.java diff --git a/hapi-fhir-checkstyle/src/checkstyle/hapi-base-checkstyle.xml b/hapi-fhir-checkstyle/src/checkstyle/hapi-base-checkstyle.xml index c74e8b7e01c..11c9b669ad4 100644 --- a/hapi-fhir-checkstyle/src/checkstyle/hapi-base-checkstyle.xml +++ b/hapi-fhir-checkstyle/src/checkstyle/hapi-base-checkstyle.xml @@ -64,4 +64,12 @@ + + + + + + + + diff --git a/hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/7_6_0/6285-batch2-reindex-version2-added.yaml b/hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/7_6_0/6285-batch2-reindex-version2-added.yaml new file mode 100644 index 00000000000..2e48e6e7bd9 --- /dev/null +++ b/hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/7_6_0/6285-batch2-reindex-version2-added.yaml @@ -0,0 +1,17 @@ +--- +type: fix +issue: 6285 +title: "Updated the Reindex Batch2 job to allow + for an additional step that will check to ensure + that no pending 'reindex' work is needed. + This was done to prevent a bug in which + value set expansion would not return all + the existing CodeSystem Concepts after + a reindex call, due to some of the concepts + being deferred to future job runs. + + As such, `$reindex` operations on CodeSystems + will no longer result in incorrect value set + expansion when such an expansion is called + 'too soon' after a $reindex operation. +" diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/BaseHapiFhirResourceDao.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/BaseHapiFhirResourceDao.java index fbd583d0090..ebe9c4aa574 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/BaseHapiFhirResourceDao.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/BaseHapiFhirResourceDao.java @@ -21,7 +21,6 @@ package ca.uhn.fhir.jpa.dao; import ca.uhn.fhir.batch2.api.IJobCoordinator; import ca.uhn.fhir.batch2.api.IJobPartitionProvider; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexAppCtx; import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; import ca.uhn.fhir.batch2.model.JobInstanceStartRequest; import ca.uhn.fhir.context.FhirVersionEnum; @@ -158,6 +157,7 @@ import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static java.util.Objects.isNull; import static org.apache.commons.lang3.StringUtils.isBlank; import static org.apache.commons.lang3.StringUtils.isNotBlank; @@ -1315,7 +1315,7 @@ public abstract class BaseHapiFhirResourceDao extends B myJobPartitionProvider.getPartitionedUrls(theRequestDetails, urls).forEach(params::addPartitionedUrl); JobInstanceStartRequest request = new JobInstanceStartRequest(); - request.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + request.setJobDefinitionId(JOB_REINDEX); request.setParameters(params); myJobCoordinator.startInstance(theRequestDetails, request); diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/JpaResourceDaoCodeSystem.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/JpaResourceDaoCodeSystem.java index 5e8f6b053bb..270238da3fb 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/JpaResourceDaoCodeSystem.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/JpaResourceDaoCodeSystem.java @@ -27,7 +27,11 @@ import ca.uhn.fhir.context.support.IValidationSupport.CodeValidationResult; import ca.uhn.fhir.context.support.LookupCodeRequest; import ca.uhn.fhir.context.support.ValidationSupportContext; import ca.uhn.fhir.i18n.Msg; +import ca.uhn.fhir.jpa.api.config.JpaStorageSettings; import ca.uhn.fhir.jpa.api.dao.IFhirResourceDaoCodeSystem; +import ca.uhn.fhir.jpa.api.dao.ReindexOutcome; +import ca.uhn.fhir.jpa.api.dao.ReindexParameters; +import ca.uhn.fhir.jpa.api.model.ReindexJobStatus; import ca.uhn.fhir.jpa.api.svc.IIdHelperService; import ca.uhn.fhir.jpa.model.cross.IBasePersistedResource; import ca.uhn.fhir.jpa.model.entity.ResourceTable; @@ -176,6 +180,47 @@ public class JpaResourceDaoCodeSystem extends BaseHapiF myTermDeferredStorageSvc.deleteCodeSystemForResource(theEntityToDelete); } + /** + * If there are more code systems to process + * than {@link JpaStorageSettings#getDeferIndexingForCodesystemsOfSize()}, + * then these codes will have their processing deferred (for a later time). + * + * This can result in future reindex steps *skipping* these code systems (if + * they're still deferred) and thus incorrect expansions resulting. + * + * So we override the reindex method for CodeSystems specifically to + * force reindex batch jobs to wait until all code systems are processed before + * moving on. + */ + @SuppressWarnings("rawtypes") + @Override + public ReindexOutcome reindex( + IResourcePersistentId thePid, + ReindexParameters theReindexParameters, + RequestDetails theRequest, + TransactionDetails theTransactionDetails) { + ReindexOutcome outcome = super.reindex(thePid, theReindexParameters, theRequest, theTransactionDetails); + + if (outcome.getWarnings().isEmpty()) { + outcome.setHasPendingWork(true); + } + return outcome; + } + + @Override + public ReindexJobStatus getReindexJobStatus() { + boolean isQueueEmpty = myTermDeferredStorageSvc.isStorageQueueEmpty(true); + + ReindexJobStatus status = new ReindexJobStatus(); + status.setHasReindexWorkPending(!isQueueEmpty); + if (status.isHasReindexWorkPending()) { + // force a run + myTermDeferredStorageSvc.saveDeferred(); + } + + return status; + } + @Override public ResourceTable updateEntity( RequestDetails theRequest, diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/TermCodeSystemStorageSvcImpl.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/TermCodeSystemStorageSvcImpl.java index af01a692a32..433f3f5d01c 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/TermCodeSystemStorageSvcImpl.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/TermCodeSystemStorageSvcImpl.java @@ -593,7 +593,7 @@ public class TermCodeSystemStorageSvcImpl implements ITermCodeSystemStorageSvc { if (theStatisticsTracker.getUpdatedConceptCount() <= myStorageSettings.getDeferIndexingForCodesystemsOfSize()) { saveConcept(conceptToAdd); Long nextConceptPid = conceptToAdd.getId(); - Validate.notNull(nextConceptPid); + Objects.requireNonNull(nextConceptPid); } else { myDeferredStorageSvc.addConceptToStorageQueue(conceptToAdd); } diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/TermDeferredStorageSvcImpl.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/TermDeferredStorageSvcImpl.java index f5e1b95d49c..47aafd066dd 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/TermDeferredStorageSvcImpl.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/TermDeferredStorageSvcImpl.java @@ -22,7 +22,6 @@ package ca.uhn.fhir.jpa.term; import ca.uhn.fhir.batch2.api.IJobCoordinator; import ca.uhn.fhir.batch2.model.JobInstance; import ca.uhn.fhir.batch2.model.JobInstanceStartRequest; -import ca.uhn.fhir.batch2.model.StatusEnum; import ca.uhn.fhir.jpa.batch.models.Batch2JobStartResponse; import ca.uhn.fhir.jpa.dao.data.ITermCodeSystemDao; import ca.uhn.fhir.jpa.dao.data.ITermCodeSystemVersionDao; @@ -79,6 +78,8 @@ public class TermDeferredStorageSvcImpl implements ITermDeferredStorageSvc, IHas private static final long SAVE_ALL_DEFERRED_WARN_MINUTES = 1; private static final long SAVE_ALL_DEFERRED_ERROR_MINUTES = 5; private boolean myAllowDeferredTasksTimeout = true; + private static final List BATCH_JOBS_TO_CARE_ABOUT = + List.of(TERM_CODE_SYSTEM_DELETE_JOB_NAME, TERM_CODE_SYSTEM_VERSION_DELETE_JOB_NAME); private final List myDeferredCodeSystemsDeletions = Collections.synchronizedList(new ArrayList<>()); private final Queue myDeferredCodeSystemVersionsDeletions = new ConcurrentLinkedQueue<>(); private final List myDeferredConcepts = Collections.synchronizedList(new ArrayList<>()); @@ -436,7 +437,7 @@ public class TermDeferredStorageSvcImpl implements ITermDeferredStorageSvc, IHas return retVal; } - private boolean isJobsExecuting() { + public boolean isJobsExecuting() { cleanseEndedJobs(); return !myJobExecutions.isEmpty(); @@ -448,15 +449,18 @@ public class TermDeferredStorageSvcImpl implements ITermDeferredStorageSvc, IHas * This is mostly a fail-safe * because "cancelled" jobs are never removed. */ - List executions = new ArrayList<>(myJobExecutions); List idsToDelete = new ArrayList<>(); - for (String id : executions) { - // TODO - might want to consider a "fetch all instances" - JobInstance instance = myJobCoordinator.getInstance(id); - if (StatusEnum.getEndedStatuses().contains(instance.getStatus())) { + for (String jobId : BATCH_JOBS_TO_CARE_ABOUT) { + List jobInstanceInEndedState = myJobCoordinator.getInstancesbyJobDefinitionIdAndEndedStatus( + jobId, + true, // ended = true (COMPLETED, FAILED, CANCELLED jobs only) + Math.max(myJobExecutions.size(), 1), // at most this many + 0); + for (JobInstance instance : jobInstanceInEndedState) { idsToDelete.add(instance.getInstanceId()); } } + for (String id : idsToDelete) { myJobExecutions.remove(id); } diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/api/ITermDeferredStorageSvc.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/api/ITermDeferredStorageSvc.java index 62fb8d8957a..6b8c3ad0dab 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/api/ITermDeferredStorageSvc.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/api/ITermDeferredStorageSvc.java @@ -78,6 +78,8 @@ public interface ITermDeferredStorageSvc { void logQueueForUnitTest(); + boolean isJobsExecuting(); + /** * Only to be used from tests - Disallow test timeouts on deferred tasks */ diff --git a/hapi-fhir-jpaserver-test-dstu3/src/test/java/ca/uhn/fhir/jpa/provider/dstu3/ResourceProviderCustomSearchParamDstu3Test.java b/hapi-fhir-jpaserver-test-dstu3/src/test/java/ca/uhn/fhir/jpa/provider/dstu3/ResourceProviderCustomSearchParamDstu3Test.java index 9dcf6439dbb..6e1511543c8 100644 --- a/hapi-fhir-jpaserver-test-dstu3/src/test/java/ca/uhn/fhir/jpa/provider/dstu3/ResourceProviderCustomSearchParamDstu3Test.java +++ b/hapi-fhir-jpaserver-test-dstu3/src/test/java/ca/uhn/fhir/jpa/provider/dstu3/ResourceProviderCustomSearchParamDstu3Test.java @@ -1,14 +1,11 @@ package ca.uhn.fhir.jpa.provider.dstu3; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexAppCtx; import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; import ca.uhn.fhir.batch2.model.JobInstance; import ca.uhn.fhir.i18n.Msg; import ca.uhn.fhir.jpa.dao.BaseHapiFhirDao; -import ca.uhn.fhir.jpa.model.entity.StorageSettings; import ca.uhn.fhir.jpa.model.entity.ResourceTable; +import ca.uhn.fhir.jpa.model.entity.StorageSettings; import ca.uhn.fhir.jpa.searchparam.SearchParameterMap; import ca.uhn.fhir.rest.api.Constants; import ca.uhn.fhir.rest.api.server.IBundleProvider; @@ -46,9 +43,11 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.jupiter.api.Assertions.fail; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.fail; @@ -202,7 +201,7 @@ public class ResourceProviderCustomSearchParamDstu3Test extends BaseResourceProv mySearchParameterDao.create(fooSp, mySrd); runInTransaction(()->{ - List allJobs = myBatch2JobHelper.findJobsByDefinition(ReindexAppCtx.JOB_REINDEX); + List allJobs = myBatch2JobHelper.findJobsByDefinition(JOB_REINDEX); assertEquals(1, allJobs.size()); assertEquals(1, allJobs.get(0).getParameters(ReindexJobParameters.class).getPartitionedUrls().size()); assertEquals("Patient?", allJobs.get(0).getParameters(ReindexJobParameters.class).getPartitionedUrls().get(0).getUrl()); diff --git a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4ComboUniqueParamTest.java b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4ComboUniqueParamTest.java index 05f4c5a5e24..b46d8b964d6 100644 --- a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4ComboUniqueParamTest.java +++ b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4ComboUniqueParamTest.java @@ -55,6 +55,7 @@ import java.util.List; import java.util.UUID; import java.util.stream.Collectors; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static ca.uhn.fhir.jpa.dao.BaseHapiFhirDao.INDEX_STATUS_INDEXED; import static ca.uhn.fhir.jpa.dao.BaseHapiFhirDao.INDEX_STATUS_INDEXING_FAILED; import static org.assertj.core.api.Assertions.assertThat; @@ -1075,7 +1076,7 @@ public class FhirResourceDaoR4ComboUniqueParamTest extends BaseComboParamsR4Test parameters.addUrl(url); } JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(parameters); Batch2JobStartResponse res = myJobCoordinator.startInstance(mySrd, startRequest); ourLog.info("Started reindex job with id {}", res.getInstanceId()); diff --git a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4IndexStorageOptimizedTest.java b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4IndexStorageOptimizedTest.java index d94b0089337..98eeb191a21 100644 --- a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4IndexStorageOptimizedTest.java +++ b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4IndexStorageOptimizedTest.java @@ -23,7 +23,7 @@ import ca.uhn.fhir.jpa.model.entity.ResourceIndexedSearchParamUri; import ca.uhn.fhir.jpa.model.entity.StorageSettings; import ca.uhn.fhir.jpa.model.util.SearchParamHash; import ca.uhn.fhir.jpa.model.util.UcumServiceUtil; -import ca.uhn.fhir.jpa.reindex.ReindexStepTest; +import ca.uhn.fhir.jpa.reindex.ReindexStepV1Test; import ca.uhn.fhir.jpa.searchparam.SearchParameterMap; import ca.uhn.fhir.jpa.test.BaseJpaR4Test; import ca.uhn.fhir.rest.param.BaseParam; @@ -57,6 +57,7 @@ import org.springframework.data.jpa.repository.JpaRepository; import java.util.List; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -312,7 +313,7 @@ public class FhirResourceDaoR4IndexStorageOptimizedTest extends BaseJpaR4Test { parameters.addUrl(url); } JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(parameters); Batch2JobStartResponse res = myJobCoordinator.startInstance(mySrd, startRequest); ourLog.info("Started reindex job with id {}", res.getInstanceId()); @@ -321,7 +322,7 @@ public class FhirResourceDaoR4IndexStorageOptimizedTest extends BaseJpaR4Test { // Additional existing tests with enabled IndexStorageOptimized @Nested - public class IndexStorageOptimizedReindexStepTest extends ReindexStepTest { + public class IndexStorageOptimizedReindexStepTestV1 extends ReindexStepV1Test { @BeforeEach void setUp() { myStorageSettings.setIndexStorageOptimized(true); diff --git a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4QueryCountTest.java b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4QueryCountTest.java index 52735147cd9..9f5ed69f7a9 100644 --- a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4QueryCountTest.java +++ b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4QueryCountTest.java @@ -2,12 +2,15 @@ package ca.uhn.fhir.jpa.dao.r4; import ca.uhn.fhir.batch2.api.IJobDataSink; import ca.uhn.fhir.batch2.api.RunOutcome; +import ca.uhn.fhir.batch2.api.StepExecutionDetails; import ca.uhn.fhir.batch2.api.VoidModel; import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; import ca.uhn.fhir.batch2.jobs.chunk.TypedPidJson; import ca.uhn.fhir.batch2.jobs.expunge.DeleteExpungeStep; +import ca.uhn.fhir.batch2.jobs.reindex.v1.ReindexStepV1; import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexStep; +import ca.uhn.fhir.batch2.model.JobInstance; +import ca.uhn.fhir.batch2.model.WorkChunk; import ca.uhn.fhir.context.FhirContext; import ca.uhn.fhir.context.support.ValidationSupportContext; import ca.uhn.fhir.context.support.ValueSetExpansionOptions; @@ -18,13 +21,13 @@ import ca.uhn.fhir.jpa.api.model.DeleteMethodOutcome; import ca.uhn.fhir.jpa.api.model.ExpungeOptions; import ca.uhn.fhir.jpa.api.model.HistoryCountModeEnum; import ca.uhn.fhir.jpa.dao.data.ISearchParamPresentDao; -import ca.uhn.fhir.jpa.reindex.ReindexTestHelper; import ca.uhn.fhir.jpa.entity.TermValueSet; import ca.uhn.fhir.jpa.entity.TermValueSetPreExpansionStatusEnum; import ca.uhn.fhir.jpa.interceptor.ForceOffsetSearchModeInterceptor; import ca.uhn.fhir.jpa.model.entity.ResourceTable; import ca.uhn.fhir.jpa.model.util.JpaConstants; import ca.uhn.fhir.jpa.provider.BaseResourceProviderR4Test; +import ca.uhn.fhir.jpa.reindex.ReindexTestHelper; import ca.uhn.fhir.jpa.search.PersistedJpaSearchFirstPageBundleProvider; import ca.uhn.fhir.jpa.searchparam.SearchParameterMap; import ca.uhn.fhir.jpa.subscription.triggering.ISubscriptionTriggeringSvc; @@ -146,7 +149,7 @@ public class FhirResourceDaoR4QueryCountTest extends BaseResourceProviderR4Test @Autowired private ISubscriptionTriggeringSvc mySubscriptionTriggeringSvc; @Autowired - private ReindexStep myReindexStep; + private ReindexStepV1 myReindexStepV1; @Autowired private DeleteExpungeStep myDeleteExpungeStep; @Autowired @@ -1018,7 +1021,6 @@ public class FhirResourceDaoR4QueryCountTest extends BaseResourceProviderR4Test // insert to: HFJ_RESOURCE, HFJ_RES_VER, HFJ_RES_LINK assertEquals(4, myCaptureQueriesListener.countInsertQueriesForCurrentThread()); assertEquals(0, myCaptureQueriesListener.countUpdateQueriesForCurrentThread()); - } @ParameterizedTest @@ -1031,7 +1033,6 @@ public class FhirResourceDaoR4QueryCountTest extends BaseResourceProviderR4Test }) public void testReindexJob_OptimizeStorage(boolean theOptimisticLock, ReindexParameters.OptimizeStorageModeEnum theOptimizeStorageModeEnum, int theExpectedSelectCount, int theExpectedUpdateCount) { // Setup - ResourceIdListWorkChunkJson data = new ResourceIdListWorkChunkJson(); IIdType patientId = createPatient(withActiveTrue()); IIdType orgId = createOrganization(withName("MY ORG")); @@ -1056,7 +1057,14 @@ public class FhirResourceDaoR4QueryCountTest extends BaseResourceProviderR4Test // execute myCaptureQueriesListener.clear(); - RunOutcome outcome = myReindexStep.doReindex(data, mock(IJobDataSink.class), "123", "456", params); + JobInstance instance = new JobInstance(); + StepExecutionDetails stepExecutionDetails = new StepExecutionDetails<>( + params, + data, + instance, + mock(WorkChunk.class) + ); + RunOutcome outcome = myReindexStepV1.run(stepExecutionDetails, mock(IJobDataSink.class)); // validate assertThat(myCaptureQueriesListener.getSelectQueriesForCurrentThread()).hasSize(theExpectedSelectCount); @@ -1064,7 +1072,6 @@ public class FhirResourceDaoR4QueryCountTest extends BaseResourceProviderR4Test assertThat(myCaptureQueriesListener.getInsertQueriesForCurrentThread()).isEmpty(); assertThat(myCaptureQueriesListener.getDeleteQueriesForCurrentThread()).isEmpty(); assertEquals(10, outcome.getRecordsProcessed()); - } @Test @@ -1095,7 +1102,14 @@ public class FhirResourceDaoR4QueryCountTest extends BaseResourceProviderR4Test // execute myCaptureQueriesListener.clear(); - RunOutcome outcome = myReindexStep.doReindex(data, mock(IJobDataSink.class), "123", "456", params); + JobInstance instance = new JobInstance(); + StepExecutionDetails stepExecutionDetails = new StepExecutionDetails<>( + params, + data, + instance, + mock(WorkChunk.class) + ); + RunOutcome outcome = myReindexStepV1.run(stepExecutionDetails, mock(IJobDataSink.class)); assertEquals(20, outcome.getRecordsProcessed()); // validate @@ -1103,10 +1117,8 @@ public class FhirResourceDaoR4QueryCountTest extends BaseResourceProviderR4Test assertEquals(0, myCaptureQueriesListener.getUpdateQueriesForCurrentThread().size()); assertEquals(0, myCaptureQueriesListener.getInsertQueriesForCurrentThread().size()); assertEquals(0, myCaptureQueriesListener.getDeleteQueriesForCurrentThread().size()); - } - public void assertNoPartitionSelectors() { List selectQueries = myCaptureQueriesListener.getSelectQueriesForCurrentThread(); for (SqlQuery next : selectQueries) { diff --git a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4SearchCustomSearchParamTest.java b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4SearchCustomSearchParamTest.java index bff403a5ea4..a9f3cc57519 100644 --- a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4SearchCustomSearchParamTest.java +++ b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4SearchCustomSearchParamTest.java @@ -1,6 +1,5 @@ package ca.uhn.fhir.jpa.dao.r4; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexAppCtx; import ca.uhn.fhir.batch2.model.JobInstance; import ca.uhn.fhir.context.RuntimeSearchParam; import ca.uhn.fhir.context.phonetic.PhoneticEncoderEnum; @@ -22,7 +21,6 @@ import ca.uhn.fhir.rest.param.NumberParam; import ca.uhn.fhir.rest.param.ReferenceOrListParam; import ca.uhn.fhir.rest.param.ReferenceParam; import ca.uhn.fhir.rest.param.StringParam; -import ca.uhn.fhir.rest.param.TokenOrListParam; import ca.uhn.fhir.rest.param.TokenParam; import ca.uhn.fhir.rest.server.exceptions.InvalidRequestException; import ca.uhn.fhir.rest.server.exceptions.UnprocessableEntityException; @@ -70,6 +68,7 @@ import org.springframework.transaction.support.TransactionTemplate; import java.util.List; import java.util.stream.Collectors; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static org.apache.commons.lang3.StringUtils.countMatches; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -441,11 +440,11 @@ public class FhirResourceDaoR4SearchCustomSearchParamTest extends BaseJpaR4Test fooSp.setXpathUsage(org.hl7.fhir.r4.model.SearchParameter.XPathUsageType.NORMAL); fooSp.setStatus(org.hl7.fhir.r4.model.Enumerations.PublicationStatus.ACTIVE); - List initialJobs = myBatch2JobHelper.findJobsByDefinition(ReindexAppCtx.JOB_REINDEX); + List initialJobs = myBatch2JobHelper.findJobsByDefinition(JOB_REINDEX); mySearchParameterDao.create(fooSp, mySrd); - List finalJobs = myBatch2JobHelper.findJobsByDefinition(ReindexAppCtx.JOB_REINDEX); + List finalJobs = myBatch2JobHelper.findJobsByDefinition(JOB_REINDEX); List newJobs = finalJobs.stream().filter(t -> !initialJobs.contains(t)).collect(Collectors.toList()); assertThat(newJobs.size()).as("number of jobs created").isEqualTo(1); } diff --git a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/tx/ReindexStepTest.java b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/tx/ReindexStepV1Test.java similarity index 92% rename from hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/tx/ReindexStepTest.java rename to hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/tx/ReindexStepV1Test.java index 69ed1e1507c..539c4590a4c 100644 --- a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/tx/ReindexStepTest.java +++ b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/tx/ReindexStepV1Test.java @@ -6,7 +6,7 @@ import ca.uhn.fhir.batch2.api.VoidModel; import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; import ca.uhn.fhir.batch2.jobs.parameters.PartitionedUrl; import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexStep; +import ca.uhn.fhir.batch2.jobs.reindex.v1.ReindexStepV1; import ca.uhn.fhir.interceptor.model.RequestPartitionId; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -26,7 +26,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) -public class ReindexStepTest { +public class ReindexStepV1Test { @Mock private HapiTransactionService myHapiTransactionService; @@ -34,7 +34,7 @@ public class ReindexStepTest { private IJobDataSink myDataSink; @InjectMocks - private ReindexStep myReindexStep; + private ReindexStepV1 myReindexStepV1; @Captor private ArgumentCaptor builderArgumentCaptor; @@ -51,7 +51,7 @@ public class ReindexStepTest { when(myHapiTransactionService.buildExecutionBuilder(any())).thenCallRealMethod(); // when - myReindexStep.doReindex(data, myDataSink, "index-id", "chunk-id", reindexJobParameters); + myReindexStepV1.doReindex(data, myDataSink, "index-id", "chunk-id", reindexJobParameters); // then assertMethodArgumentRequestPartitionId(expectedPartitionId); diff --git a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/provider/r4/ResourceProviderCustomSearchParamR4Test.java b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/provider/r4/ResourceProviderCustomSearchParamR4Test.java index 987f3430873..0bee635ccc0 100644 --- a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/provider/r4/ResourceProviderCustomSearchParamR4Test.java +++ b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/provider/r4/ResourceProviderCustomSearchParamR4Test.java @@ -1,8 +1,5 @@ package ca.uhn.fhir.jpa.provider.r4; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexAppCtx; import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; import ca.uhn.fhir.batch2.model.JobInstance; import ca.uhn.fhir.i18n.Msg; @@ -60,9 +57,11 @@ import java.util.List; import java.util.Map; import java.util.stream.Collectors; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.jupiter.api.Assertions.fail; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.fail; @@ -244,7 +243,7 @@ public class ResourceProviderCustomSearchParamR4Test extends BaseResourceProvide runInTransaction(() -> { myBatch2JobHelper.forceRunMaintenancePass(); - List allJobs = myBatch2JobHelper.findJobsByDefinition(ReindexAppCtx.JOB_REINDEX); + List allJobs = myBatch2JobHelper.findJobsByDefinition(JOB_REINDEX); assertEquals(1, allJobs.size()); assertEquals(1, allJobs.get(0).getParameters(ReindexJobParameters.class).getPartitionedUrls().size()); assertEquals("Patient?", allJobs.get(0).getParameters(ReindexJobParameters.class).getPartitionedUrls().get(0).getUrl()); diff --git a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexStepTest.java b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexStepV1Test.java similarity index 80% rename from hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexStepTest.java rename to hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexStepV1Test.java index d7ed24e6789..8075789db8d 100644 --- a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexStepTest.java +++ b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexStepV1Test.java @@ -2,10 +2,13 @@ package ca.uhn.fhir.jpa.reindex; import ca.uhn.fhir.batch2.api.IJobDataSink; import ca.uhn.fhir.batch2.api.RunOutcome; +import ca.uhn.fhir.batch2.api.StepExecutionDetails; import ca.uhn.fhir.batch2.api.VoidModel; import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexStep; +import ca.uhn.fhir.batch2.jobs.reindex.v1.ReindexStepV1; +import ca.uhn.fhir.batch2.model.JobInstance; +import ca.uhn.fhir.batch2.model.WorkChunk; import ca.uhn.fhir.jpa.api.config.JpaStorageSettings; import ca.uhn.fhir.jpa.model.entity.ResourceTable; import ca.uhn.fhir.jpa.test.BaseJpaR4Test; @@ -25,13 +28,14 @@ import static ca.uhn.fhir.jpa.dao.BaseHapiFhirDao.INDEX_STATUS_INDEXED; import static ca.uhn.fhir.jpa.dao.BaseHapiFhirDao.INDEX_STATUS_INDEXING_FAILED; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; -public class ReindexStepTest extends BaseJpaR4Test { +public class ReindexStepV1Test extends BaseJpaR4Test { @Autowired - private ReindexStep myReindexStep; + private ReindexStepV1 myReindexStepV1; @Mock private IJobDataSink myDataSink; @@ -46,9 +50,7 @@ public class ReindexStepTest extends BaseJpaR4Test { @Test public void testReindex_NoActionNeeded() { - // Setup - Long id0 = createPatient(withActiveTrue(), withFamily("SIMPSON")).getIdPartAsLong(); Long id1 = createPatient(withActiveTrue(), withFamily("FLANDERS")).getIdPartAsLong(); @@ -57,9 +59,19 @@ public class ReindexStepTest extends BaseJpaR4Test { data.addTypedPid("Patient", id1); // Execute - + ReindexJobParameters params = new ReindexJobParameters(); myCaptureQueriesListener.clear(); - RunOutcome outcome = myReindexStep.doReindex(data, myDataSink, "index-id", "chunk-id", new ReindexJobParameters()); + JobInstance instance = new JobInstance(); + instance.setInstanceId("index-id"); + WorkChunk chunk = new WorkChunk(); + chunk.setId("chunk-id"); + StepExecutionDetails stepExecutionDetails = new StepExecutionDetails<>( + params, + data, + instance, + chunk + ); + RunOutcome outcome = myReindexStepV1.run(stepExecutionDetails, myDataSink); // Verify assertEquals(2, outcome.getRecordsProcessed()); @@ -72,12 +84,9 @@ public class ReindexStepTest extends BaseJpaR4Test { assertEquals(0, myCaptureQueriesListener.getRollbackCount()); } - @Test public void testReindex_NoActionNeeded_IndexMissingFieldsEnabled() { - // Setup - myStorageSettings.setIndexMissingFields(JpaStorageSettings.IndexEnabledEnum.ENABLED); Long id0 = createPatient(withActiveTrue(), withFamily("SIMPSON")).getIdPartAsLong(); @@ -88,9 +97,16 @@ public class ReindexStepTest extends BaseJpaR4Test { data.addTypedPid("Patient", id1); // Execute - + ReindexJobParameters params = new ReindexJobParameters(); myCaptureQueriesListener.clear(); - RunOutcome outcome = myReindexStep.doReindex(data, myDataSink, "index-id", "chunk-id", new ReindexJobParameters()); + JobInstance instance = new JobInstance(); + StepExecutionDetails stepExecutionDetails = new StepExecutionDetails<>( + params, + data, + instance, + mock(WorkChunk.class) + ); + RunOutcome outcome = myReindexStepV1.run(stepExecutionDetails, myDataSink); // Verify assertEquals(2, outcome.getRecordsProcessed()); @@ -121,9 +137,16 @@ public class ReindexStepTest extends BaseJpaR4Test { }); // Execute - + ReindexJobParameters params = new ReindexJobParameters(); myCaptureQueriesListener.clear(); - RunOutcome outcome = myReindexStep.doReindex(data, myDataSink, "index-id", "chunk-id", new ReindexJobParameters()); + JobInstance instance = new JobInstance(); + StepExecutionDetails stepExecutionDetails = new StepExecutionDetails<>( + params, + data, + instance, + mock(WorkChunk.class) + ); + RunOutcome outcome = myReindexStepV1.run(stepExecutionDetails, myDataSink); // Verify assertEquals(2, outcome.getRecordsProcessed()); @@ -136,12 +159,9 @@ public class ReindexStepTest extends BaseJpaR4Test { assertEquals(0, myCaptureQueriesListener.getRollbackCount()); } - @Test public void testReindex_IndexesAddedAndRemoved_IndexMissingFieldsEnabled() { - // Setup - myStorageSettings.setIndexMissingFields(JpaStorageSettings.IndexEnabledEnum.ENABLED); boolean markResourcesForReindexingUponSearchParameterChange = myStorageSettings.isMarkResourcesForReindexingUponSearchParameterChange(); myStorageSettings.setMarkResourcesForReindexingUponSearchParameterChange(false); // if this were true, it would set up a lot of reindex jobs extraneous to the one we're trying to test @@ -189,9 +209,16 @@ public class ReindexStepTest extends BaseJpaR4Test { mySearchParamRegistry.forceRefresh(); // Execute - + ReindexJobParameters params = new ReindexJobParameters(); myCaptureQueriesListener.clear(); - RunOutcome outcome = myReindexStep.doReindex(data, myDataSink, "index-id", "chunk-id", new ReindexJobParameters()); + JobInstance instance = new JobInstance(); + StepExecutionDetails stepExecutionDetails = new StepExecutionDetails<>( + params, + data, + instance, + mock(WorkChunk.class) + ); + RunOutcome outcome = myReindexStepV1.run(stepExecutionDetails, myDataSink); // Verify assertEquals(2, outcome.getRecordsProcessed()); @@ -207,9 +234,7 @@ public class ReindexStepTest extends BaseJpaR4Test { @Test public void testReindex_OneResourceReindexFailedButOthersSucceeded() { - // Setup - Long id0 = createPatient(withActiveTrue(), withFamily("SIMPSON")).getIdPartAsLong(); Long id1 = createPatient(withActiveTrue(), withFamily("FLANDERS")).getIdPartAsLong(); Long idPatientToInvalidate = createPatient().getIdPartAsLong(); @@ -234,9 +259,19 @@ public class ReindexStepTest extends BaseJpaR4Test { }); // Execute - + ReindexJobParameters params = new ReindexJobParameters(); myCaptureQueriesListener.clear(); - RunOutcome outcome = myReindexStep.doReindex(data, myDataSink, "index-id", "chunk-id", new ReindexJobParameters()); + JobInstance instance = new JobInstance(); + instance.setInstanceId("index-id"); + WorkChunk workChunk = new WorkChunk(); + workChunk.setId("workid"); + StepExecutionDetails stepExecutionDetails = new StepExecutionDetails<>( + params, + data, + instance, + workChunk + ); + RunOutcome outcome = myReindexStepV1.run(stepExecutionDetails, myDataSink); // Verify assertEquals(4, outcome.getRecordsProcessed()); diff --git a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexJobTest.java b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexTaskTest.java similarity index 95% rename from hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexJobTest.java rename to hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexTaskTest.java index fa9d93fb4ab..c04a40f1ac7 100644 --- a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexJobTest.java +++ b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexTaskTest.java @@ -2,7 +2,6 @@ package ca.uhn.fhir.jpa.reindex; import ca.uhn.fhir.batch2.api.IJobCoordinator; import ca.uhn.fhir.batch2.api.IJobPersistence; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexAppCtx; import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; import ca.uhn.fhir.batch2.model.JobInstance; import ca.uhn.fhir.batch2.model.JobInstanceStartRequest; @@ -39,6 +38,7 @@ import java.util.Date; import java.util.List; import java.util.stream.Stream; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -47,7 +47,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @SuppressWarnings("SqlDialectInspection") -public class ReindexJobTest extends BaseJpaR4Test { +public class ReindexTaskTest extends BaseJpaR4Test { @Autowired private IJobCoordinator myJobCoordinator; @@ -101,7 +101,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // execute JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters( new ReindexJobParameters() .setOptimizeStorage(ReindexParameters.OptimizeStorageModeEnum.CURRENT_VERSION) @@ -158,7 +158,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // execute JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters( new ReindexJobParameters() .setOptimizeStorage(ReindexParameters.OptimizeStorageModeEnum.ALL_VERSIONS) @@ -217,7 +217,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // execute JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters( new ReindexJobParameters() .setOptimizeStorage(ReindexParameters.OptimizeStorageModeEnum.ALL_VERSIONS) @@ -252,7 +252,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // execute JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters( new ReindexJobParameters() .setOptimizeStorage(ReindexParameters.OptimizeStorageModeEnum.CURRENT_VERSION) @@ -294,7 +294,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // execute JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(parameters); Batch2JobStartResponse res = myJobCoordinator.startInstance(mySrd, startRequest); myBatch2JobHelper.awaitJobCompletion(res); @@ -325,7 +325,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // execute JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(parameters); Batch2JobStartResponse res = myJobCoordinator.startInstance(startRequest); JobInstance jobInstance = myBatch2JobHelper.awaitJobCompletion(res); @@ -356,7 +356,7 @@ public class ReindexJobTest extends BaseJpaR4Test { parameters.addUrl("Observation?status=final"); JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(parameters); Batch2JobStartResponse res = myJobCoordinator.startInstance(mySrd, startRequest); myBatch2JobHelper.awaitJobCompletion(res); @@ -387,7 +387,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // execute JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(new ReindexJobParameters()); Batch2JobStartResponse startResponse = myJobCoordinator.startInstance(mySrd, startRequest); myBatch2JobHelper.awaitJobCompletion(startResponse); @@ -406,7 +406,7 @@ public class ReindexJobTest extends BaseJpaR4Test { DaoMethodOutcome searchParameter = myReindexTestHelper.createUniqueCodeSearchParameter(); JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(new ReindexJobParameters()); Batch2JobStartResponse startResponse = myJobCoordinator.startInstance(new SystemRequestDetails(), startRequest); JobInstance myJob = myBatch2JobHelper.awaitJobCompletion(startResponse); @@ -436,7 +436,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // Run a reindex JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(new ReindexJobParameters()); Batch2JobStartResponse startResponse = myJobCoordinator.startInstance(new SystemRequestDetails(), startRequest); JobInstance myJob = myBatch2JobHelper.awaitJobCompletion(startResponse.getInstanceId(), 999); @@ -469,7 +469,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // Run a reindex JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(new ReindexJobParameters()); Batch2JobStartResponse startResponse = myJobCoordinator.startInstance(new SystemRequestDetails(), startRequest); JobInstance myJob = myBatch2JobHelper.awaitJobCompletion(startResponse.getInstanceId(), 999); @@ -500,7 +500,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // execute JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(new ReindexJobParameters()); Batch2JobStartResponse startResponse = myJobCoordinator.startInstance(mySrd, startRequest); JobInstance outcome = myBatch2JobHelper.awaitJobCompletion(startResponse); @@ -528,7 +528,7 @@ public class ReindexJobTest extends BaseJpaR4Test { // execute JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(new ReindexJobParameters()); Batch2JobStartResponse startResponse = myJobCoordinator.startInstance(new SystemRequestDetails(), startRequest); JobInstance outcome = myBatch2JobHelper.awaitJobFailure(startResponse); @@ -541,7 +541,7 @@ public class ReindexJobTest extends BaseJpaR4Test { @Test public void testReindex_withReindexingUponSearchParameterChangeEnabled_reindexJobCompleted() { - List jobInstances = myJobPersistence.fetchInstancesByJobDefinitionId(ReindexAppCtx.JOB_REINDEX, 10, 0); + List jobInstances = myJobPersistence.fetchInstancesByJobDefinitionId(JOB_REINDEX, 10, 0); assertThat(jobInstances).isEmpty(); // make sure the resources auto-reindex after the search parameter update is enabled @@ -552,7 +552,7 @@ public class ReindexJobTest extends BaseJpaR4Test { myReindexTestHelper.createCodeSearchParameter(); // check that reindex job was created - jobInstances = myJobPersistence.fetchInstancesByJobDefinitionId(ReindexAppCtx.JOB_REINDEX, 10, 0); + jobInstances = myJobPersistence.fetchInstancesByJobDefinitionId(JOB_REINDEX, 10, 0); assertThat(jobInstances).hasSize(1); // check that the job is completed (not stuck in QUEUED status) diff --git a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexJobWithPartitioningTest.java b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexTaskWithPartitioningTest.java similarity index 97% rename from hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexJobWithPartitioningTest.java rename to hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexTaskWithPartitioningTest.java index 8a643006439..35657c3ab47 100644 --- a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexJobWithPartitioningTest.java +++ b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/reindex/ReindexTaskWithPartitioningTest.java @@ -1,9 +1,8 @@ package ca.uhn.fhir.jpa.reindex; import ca.uhn.fhir.batch2.api.IJobCoordinator; -import ca.uhn.fhir.batch2.jobs.parameters.PartitionedUrlJobParameters; import ca.uhn.fhir.batch2.jobs.parameters.PartitionedUrl; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexAppCtx; +import ca.uhn.fhir.batch2.jobs.parameters.PartitionedUrlJobParameters; import ca.uhn.fhir.batch2.model.JobInstance; import ca.uhn.fhir.batch2.model.JobInstanceStartRequest; import ca.uhn.fhir.interceptor.model.RequestPartitionId; @@ -25,9 +24,10 @@ import org.springframework.beans.factory.annotation.Autowired; import java.util.List; import java.util.stream.Stream; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static org.assertj.core.api.Assertions.assertThat; @TestInstance(TestInstance.Lifecycle.PER_CLASS) -public class ReindexJobWithPartitioningTest extends BaseJpaR4Test { +public class ReindexTaskWithPartitioningTest extends BaseJpaR4Test { @Autowired private IJobCoordinator myJobCoordinator; @@ -133,7 +133,7 @@ public class ReindexJobWithPartitioningTest extends BaseJpaR4Test { // execute JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(parameters); Batch2JobStartResponse res = myJobCoordinator.startInstance(new SystemRequestDetails(), startRequest); JobInstance jobInstance = myBatch2JobHelper.awaitJobCompletion(res); diff --git a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/term/ValueSetExpansionR4Test.java b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/term/ValueSetExpansionR4Test.java index 1f9f229c72a..d6f5039d129 100644 --- a/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/term/ValueSetExpansionR4Test.java +++ b/hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/term/ValueSetExpansionR4Test.java @@ -1,5 +1,8 @@ package ca.uhn.fhir.jpa.term; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils; +import ca.uhn.fhir.batch2.model.JobInstanceStartRequest; import ca.uhn.fhir.context.FhirVersionEnum; import ca.uhn.fhir.context.RuntimeSearchParam; import ca.uhn.fhir.context.support.ConceptValidationOptions; @@ -9,6 +12,7 @@ import ca.uhn.fhir.i18n.Msg; import ca.uhn.fhir.jpa.api.config.JpaStorageSettings; import ca.uhn.fhir.jpa.api.dao.DaoRegistry; import ca.uhn.fhir.jpa.api.dao.IFhirResourceDaoValueSet; +import ca.uhn.fhir.jpa.batch.models.Batch2JobStartResponse; import ca.uhn.fhir.jpa.entity.TermCodeSystem; import ca.uhn.fhir.jpa.entity.TermCodeSystemVersion; import ca.uhn.fhir.jpa.entity.TermConcept; @@ -25,7 +29,10 @@ import ca.uhn.fhir.jpa.term.api.ITermReadSvc; import ca.uhn.fhir.jpa.term.custom.CustomTerminologySet; import ca.uhn.fhir.jpa.util.SqlQuery; import ca.uhn.fhir.jpa.util.ValueSetTestUtil; +import ca.uhn.fhir.parser.IParser; import ca.uhn.fhir.rest.api.server.IBundleProvider; +import ca.uhn.fhir.rest.api.server.RequestDetails; +import ca.uhn.fhir.rest.api.server.SystemRequestDetails; import ca.uhn.fhir.rest.server.exceptions.InternalErrorException; import ca.uhn.fhir.rest.server.exceptions.InvalidRequestException; import com.google.common.collect.Lists; @@ -51,10 +58,16 @@ import org.springframework.transaction.support.TransactionCallbackWithoutResult; import org.springframework.transaction.support.TransactionTemplate; import java.io.IOException; +import java.time.Duration; +import java.time.temporal.ChronoUnit; +import java.util.Collection; +import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.stream.Collectors; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static ca.uhn.fhir.util.HapiExtensions.EXT_VALUESET_EXPANSION_MESSAGE; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; @@ -2081,7 +2094,172 @@ public class ValueSetExpansionR4Test extends BaseTermR4Test implements IValueSet outcome = myValueSetDao.validateCode(vs.getUrlElement(), null, new StringType("A"), cs.getUrlElement(), null, null, null, mySrd); assertEquals(false, outcome.isOk()); assertThat(outcome.getMessage()).contains("Code validation occurred using a ValueSet expansion that was pre-calculated"); - } + @Test + public void reindexCodeSystems_withDeferredCodeSystems_reindexesAllCodeSystems() { + // setup + int deferredIndexingDefault = myStorageSettings.getDeferIndexingForCodesystemsOfSize(); + + try { + /** + * The deferred count must be less than the number of + * concepts we are going to be uploading. + * That way, when we do the reindex, it will defer + * the additional code systems for a later job run. + * + * See {@link TermCodeSystemStorageSvcImpl#addConceptInHierarchy(TermCodeSystemVersion, Collection, TermConcept, UploadStatistics, Map, int)} + * + * Our CodeSystem below only has 6 Concepts to add. + * So we'll set the deferred count to 3 (so 3 will be deferred) + */ + myStorageSettings.setDeferIndexingForCodesystemsOfSize(3); + /* + * We're also setting our retry delay to a short timeframe + * so this test doesn't run too long. + */ + ReindexUtils.setRetryDelay(Duration.of(300, ChronoUnit.MILLIS)); + + IParser parser = myFhirContext.newJsonParser(); + + RequestDetails rq = new SystemRequestDetails(); + CodeSystem cs; + ValueSet vs; + String csStr; + { + String vsStr = """ + { + "resourceType": "ValueSet", + "id": "0447bffa-01fa-4405-828a-96192e74a5d8", + "meta": { + "versionId": "2", + "lastUpdated": "2024-04-09T15:06:24.025+00:00", + "source": "#f4491e490a6a2900" + }, + "url": "https://health.gov.on.ca/idms/fhir/ValueSet/IDMS-Submission-Types", + "version": "1.0.0", + "name": "IDMS-SUBMISSION-TYPES", + "title": "IDMS Submission Types", + "status": "active", + "experimental": false, + "date": "2023-09-28", + "publisher": "IDMS", + "description": "List of Submission Types", + "compose": { + "include": [ + { + "system": "https://health.gov.on.ca/idms/fhir/CodeSystem/Internal-Submission-Types" + } + ] + } + } + """; + vs = parser.parseResource(ValueSet.class, vsStr); + csStr = """ + { + "resourceType": "CodeSystem", + "id": "d9acd5b8-9533-4fa1-bb70-b4380957a8c3", + "meta": { + "versionId": "14", + "lastUpdated": "2024-06-03T17:49:56.580+00:00", + "source": "#261a82258b0978a8" + }, + "url": "https://health.gov.on.ca/idms/fhir/CodeSystem/Internal-Submission-Types", + "version": "1.0.0", + "name": "IDMS-Internal-Submission-Types", + "status": "active", + "date": "2023-09-07", + "publisher": "IDMS", + "description": "This contains a lists of codes Submission Type Codes.", + "content": "complete", + "concept": [ + { + "code": "SUB-BRAND-PRODUCT", + "display": "New Brand Product (Non-New Chemical Entity)" + }, + { + "code": "SUB-CLASSIFICATION", + "display": "Classification Change" + }, + { + "code": "SUB-CLINICIAN-LED-SUBMISSIONS", + "display": "Clinician-Led Submissions" + }, + { + "code": "SUB-DELISTING", + "display": "Delisting" + }, + { + "code": "SUB-DIN-CHANGE", + "display": "Drug Identification Number (DIN) Change" + }, + { + "code": "SUB-DISTRIBUTOR", + "display": "Distributor Change" + } + ] + } + """; + cs = parser.parseResource(CodeSystem.class, csStr); + } + + // create our ValueSet + myValueSetDao.update(vs, rq); + + // and the code system + myCodeSystemDao.update(cs, rq); + + // sanity check to make sure our code system was actually created + SearchParameterMap spMap = new SearchParameterMap(); + spMap.setLoadSynchronous(true); + IBundleProvider bp = myCodeSystemDao.search(spMap, rq); + assertEquals(1, bp.getAllResources().size()); + IBaseResource baseResource = bp.getAllResources().get(0); + CodeSystem cssaved; + if (baseResource instanceof CodeSystem saved) { + cssaved = saved; + } else { + fail("Should be a code system"); + return; + } + assertEquals(cs.getConcept().size(), cssaved.getConcept().size()); + + // test + // perform the reindex (we'll only target the CodeSystem here) + ReindexJobParameters params = new ReindexJobParameters(); + params.addUrl("CodeSystem?"); + JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); + startRequest.setJobDefinitionId(JOB_REINDEX); + startRequest.setParameters(params); + + // and wait for it to complete + Batch2JobStartResponse response = myJobCoordinator.startInstance(rq, startRequest); + myBatch2JobHelper.awaitJobCompletion(response); + + // verify by doing the value expansion + ValueSetExpansionOptions options = new ValueSetExpansionOptions(); + options.setCount(200); // this is way more than exist, so it's ok + ValueSet expanded = myValueSetDao.expand(vs, options); + assertNotNull(expanded); + + /* + * If the reindex was performed correctly, the expanded ValueSet + * should contain all the CodeSystem concepts that we originally + * uploaded (and nothing else). + */ + HashSet all = new HashSet<>(); + for (CodeSystem.ConceptDefinitionComponent set : cs.getConcept()) { + all.add(set.getCode()); + } + for (ValueSet.ValueSetExpansionContainsComponent v : expanded.getExpansion().getContains()) { + all.remove(v.getCode()); + } + assertTrue(all.isEmpty(), String.join(", ", all)); + assertEquals(cs.getConcept().size(), expanded.getExpansion().getTotal()); + } finally { + // set back to standard values + myStorageSettings.setDeferIndexingForCodesystemsOfSize(deferredIndexingDefault); + ReindexUtils.setRetryDelay(null); + } + } } diff --git a/hapi-fhir-jpaserver-test-r5/src/test/java/ca/uhn/fhir/jpa/dao/r5/DuplicateIndexR5Test.java b/hapi-fhir-jpaserver-test-r5/src/test/java/ca/uhn/fhir/jpa/dao/r5/DuplicateIndexR5Test.java index d88d136d089..2a4387b4597 100644 --- a/hapi-fhir-jpaserver-test-r5/src/test/java/ca/uhn/fhir/jpa/dao/r5/DuplicateIndexR5Test.java +++ b/hapi-fhir-jpaserver-test-r5/src/test/java/ca/uhn/fhir/jpa/dao/r5/DuplicateIndexR5Test.java @@ -1,6 +1,5 @@ package ca.uhn.fhir.jpa.dao.r5; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexAppCtx; import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; import ca.uhn.fhir.batch2.model.JobInstanceStartRequest; import ca.uhn.fhir.jpa.batch.models.Batch2JobStartResponse; @@ -19,6 +18,7 @@ import org.hl7.fhir.r5.model.Reference; import org.hl7.fhir.r5.model.SearchParameter; import org.junit.jupiter.api.Test; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static org.junit.jupiter.api.Assertions.assertEquals; public class DuplicateIndexR5Test extends BaseJpaR5Test { @@ -149,7 +149,7 @@ public class DuplicateIndexR5Test extends BaseJpaR5Test { ReindexJobParameters parameters = new ReindexJobParameters(); parameters.addUrl("Patient?"); JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); startRequest.setParameters(parameters); Batch2JobStartResponse res = myJobCoordinator.startInstance(mySrd, startRequest); myBatch2JobHelper.awaitJobCompletion(res.getInstanceId()); diff --git a/hapi-fhir-jpaserver-test-r5/src/test/java/ca/uhn/fhir/jpa/provider/r5/ResourceProviderR5Test.java b/hapi-fhir-jpaserver-test-r5/src/test/java/ca/uhn/fhir/jpa/provider/r5/ResourceProviderR5Test.java index f6f26c10fc9..4ee5ea481c4 100644 --- a/hapi-fhir-jpaserver-test-r5/src/test/java/ca/uhn/fhir/jpa/provider/r5/ResourceProviderR5Test.java +++ b/hapi-fhir-jpaserver-test-r5/src/test/java/ca/uhn/fhir/jpa/provider/r5/ResourceProviderR5Test.java @@ -1,7 +1,6 @@ package ca.uhn.fhir.jpa.provider.r5; import ca.uhn.fhir.batch2.jobs.parameters.PartitionedUrl; -import ca.uhn.fhir.batch2.jobs.reindex.ReindexAppCtx; import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; import ca.uhn.fhir.batch2.model.JobInstanceStartRequest; import ca.uhn.fhir.interceptor.model.RequestPartitionId; @@ -67,6 +66,7 @@ import java.util.List; import java.util.Set; import java.util.stream.Collectors; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static org.apache.commons.lang3.StringUtils.leftPad; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -374,7 +374,7 @@ public class ResourceProviderR5Test extends BaseResourceProviderR5Test { ReindexJobParameters jobParameters = new ReindexJobParameters(); jobParameters.addPartitionedUrl(new PartitionedUrl().setRequestPartitionId(RequestPartitionId.allPartitions())); JobInstanceStartRequest request = new JobInstanceStartRequest(); - request.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + request.setJobDefinitionId(JOB_REINDEX); request.setParameters(jobParameters); Batch2JobStartResponse response = myJobCoordinator.startInstance(new SystemRequestDetails(), request); diff --git a/hapi-fhir-jpaserver-test-utilities/src/main/java/ca/uhn/fhir/jpa/test/BaseJpaR4Test.java b/hapi-fhir-jpaserver-test-utilities/src/main/java/ca/uhn/fhir/jpa/test/BaseJpaR4Test.java index 2fb5a9b87f6..8b98ddfe30e 100644 --- a/hapi-fhir-jpaserver-test-utilities/src/main/java/ca/uhn/fhir/jpa/test/BaseJpaR4Test.java +++ b/hapi-fhir-jpaserver-test-utilities/src/main/java/ca/uhn/fhir/jpa/test/BaseJpaR4Test.java @@ -19,6 +19,7 @@ */ package ca.uhn.fhir.jpa.test; +import ca.uhn.fhir.batch2.api.IJobCoordinator; import ca.uhn.fhir.batch2.api.IJobMaintenanceService; import ca.uhn.fhir.batch2.jobs.export.BulkDataExportProvider; import ca.uhn.fhir.context.FhirContext; @@ -559,6 +560,8 @@ public abstract class BaseJpaR4Test extends BaseJpaTest implements ITestDataBuil @Autowired protected IJobMaintenanceService myJobMaintenanceService; + @Autowired + protected IJobCoordinator myJobCoordinator; @RegisterExtension private final PreventDanglingInterceptorsExtension myPreventDanglingInterceptorsExtension = new PreventDanglingInterceptorsExtension(()-> myInterceptorRegistry); diff --git a/hapi-fhir-jpaserver-test-utilities/src/main/java/ca/uhn/fhir/jpa/test/PatientReindexTestHelper.java b/hapi-fhir-jpaserver-test-utilities/src/main/java/ca/uhn/fhir/jpa/test/PatientReindexTestHelper.java index 346ecfc519f..bb063387b89 100644 --- a/hapi-fhir-jpaserver-test-utilities/src/main/java/ca/uhn/fhir/jpa/test/PatientReindexTestHelper.java +++ b/hapi-fhir-jpaserver-test-utilities/src/main/java/ca/uhn/fhir/jpa/test/PatientReindexTestHelper.java @@ -38,6 +38,7 @@ import org.junit.jupiter.params.provider.Arguments; import java.util.List; import java.util.stream.Stream; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.fail; @@ -170,7 +171,7 @@ public class PatientReindexTestHelper { private JobInstanceStartRequest createPatientReindexRequest(int theBatchSize) { JobInstanceStartRequest startRequest = new JobInstanceStartRequest(); - startRequest.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + startRequest.setJobDefinitionId(JOB_REINDEX); ReindexJobParameters reindexJobParameters = new ReindexJobParameters(); reindexJobParameters.setBatchSize(Math.max(theBatchSize,1)); diff --git a/hapi-fhir-jpaserver-test-utilities/src/test/java/ca/uhn/fhir/jpa/term/TermDeferredStorageSvcImplTest.java b/hapi-fhir-jpaserver-test-utilities/src/test/java/ca/uhn/fhir/jpa/term/TermDeferredStorageSvcImplTest.java index eed255e7842..2365300b65a 100644 --- a/hapi-fhir-jpaserver-test-utilities/src/test/java/ca/uhn/fhir/jpa/term/TermDeferredStorageSvcImplTest.java +++ b/hapi-fhir-jpaserver-test-utilities/src/test/java/ca/uhn/fhir/jpa/term/TermDeferredStorageSvcImplTest.java @@ -17,11 +17,15 @@ import org.springframework.test.util.ReflectionTestUtils; import org.springframework.transaction.PlatformTransactionManager; import java.util.ArrayList; +import java.util.List; import java.util.Optional; +import static ca.uhn.fhir.batch2.jobs.termcodesystem.TermCodeSystemJobConfig.TERM_CODE_SYSTEM_DELETE_JOB_NAME; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.same; import static org.mockito.Mockito.times; @@ -67,8 +71,15 @@ public class TermDeferredStorageSvcImplTest { ReflectionTestUtils.setField(mySvc, "myJobExecutions", mockExecutions); - when(myJobCoordinator.getInstance(eq(jobId))) - .thenReturn(instance); + when(myJobCoordinator.getInstancesbyJobDefinitionIdAndEndedStatus( + eq(TERM_CODE_SYSTEM_DELETE_JOB_NAME), + eq(true), + anyInt(), + eq(0) + )) + .thenReturn(List.of()) // first nothing + .thenReturn(List.of(instance)); // then the list with the instance + assertFalse(mySvc.isStorageQueueEmpty(true)); instance.setStatus(StatusEnum.COMPLETED); assertTrue(mySvc.isStorageQueueEmpty(true)); diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexAppCtx.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexAppCtx.java index a49d054bf39..ba335b1f60b 100644 --- a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexAppCtx.java +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexAppCtx.java @@ -21,69 +21,37 @@ package ca.uhn.fhir.batch2.jobs.reindex; import ca.uhn.fhir.batch2.api.IJobCoordinator; import ca.uhn.fhir.batch2.api.IJobPartitionProvider; -import ca.uhn.fhir.batch2.api.IJobStepWorker; -import ca.uhn.fhir.batch2.api.VoidModel; -import ca.uhn.fhir.batch2.jobs.chunk.ChunkRangeJson; -import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; -import ca.uhn.fhir.batch2.jobs.parameters.UrlListValidator; -import ca.uhn.fhir.batch2.jobs.step.GenerateRangeChunksStep; -import ca.uhn.fhir.batch2.jobs.step.LoadIdsStep; -import ca.uhn.fhir.batch2.model.JobDefinition; +import ca.uhn.fhir.batch2.jobs.reindex.svcs.ReindexJobService; +import ca.uhn.fhir.batch2.jobs.reindex.v1.ReindexV1Config; +import ca.uhn.fhir.batch2.jobs.reindex.v2.ReindexV2Config; import ca.uhn.fhir.context.FhirContext; -import ca.uhn.fhir.jpa.api.svc.IBatch2DaoSvc; -import ca.uhn.fhir.rest.server.provider.ProviderConstants; +import ca.uhn.fhir.jpa.api.dao.DaoRegistry; +import ca.uhn.fhir.jpa.api.dao.IFhirSystemDao; +import ca.uhn.fhir.jpa.api.svc.IIdHelperService; +import ca.uhn.fhir.jpa.dao.tx.HapiTransactionService; +import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; @Configuration +@Import({ReindexV1Config.class, ReindexV2Config.class}) public class ReindexAppCtx { - public static final String JOB_REINDEX = "REINDEX"; + @Autowired + private HapiTransactionService myHapiTransactionService; - @Bean - public JobDefinition reindexJobDefinition(IBatch2DaoSvc theBatch2DaoSvc) { - return JobDefinition.newBuilder() - .setJobDefinitionId(JOB_REINDEX) - .setJobDescription("Reindex resources") - .setJobDefinitionVersion(1) - .setParametersType(ReindexJobParameters.class) - .setParametersValidator(reindexJobParametersValidator(theBatch2DaoSvc)) - .gatedExecution() - .addFirstStep( - "generate-ranges", - "Generate data ranges to reindex", - ChunkRangeJson.class, - reindexGenerateRangeChunksStep()) - .addIntermediateStep( - "load-ids", - "Load IDs of resources to reindex", - ResourceIdListWorkChunkJson.class, - reindexLoadIdsStep(theBatch2DaoSvc)) - .addLastStep("reindex", "Perform the resource reindex", reindexStep()) - .build(); - } + @Autowired + private IFhirSystemDao mySystemDao; - @Bean - public IJobStepWorker reindexGenerateRangeChunksStep() { - return new GenerateRangeChunksStep<>(); - } + @Autowired + private DaoRegistry myRegistry; - @Bean - public IJobStepWorker reindexLoadIdsStep( - IBatch2DaoSvc theBatch2DaoSvc) { - return new LoadIdsStep<>(theBatch2DaoSvc); - } + @Autowired + private IIdHelperService> myIdHelperService; - @Bean - public ReindexJobParametersValidator reindexJobParametersValidator(IBatch2DaoSvc theBatch2DaoSvc) { - return new ReindexJobParametersValidator( - new UrlListValidator(ProviderConstants.OPERATION_REINDEX, theBatch2DaoSvc)); - } - - @Bean - public ReindexStep reindexStep() { - return new ReindexStep(); - } + /* Shared services */ @Bean public ReindexProvider reindexProvider( @@ -92,4 +60,9 @@ public class ReindexAppCtx { IJobPartitionProvider theJobPartitionHandler) { return new ReindexProvider(theFhirContext, theJobCoordinator, theJobPartitionHandler); } + + @Bean + public ReindexJobService jobService() { + return new ReindexJobService(myRegistry); + } } diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexProvider.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexProvider.java index 5889a75d4e9..a8f8e08707c 100644 --- a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexProvider.java +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexProvider.java @@ -44,6 +44,7 @@ import java.util.stream.Collectors; import static ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters.OPTIMIZE_STORAGE; import static ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters.REINDEX_SEARCH_PARAMETERS; +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; public class ReindexProvider { @@ -127,7 +128,7 @@ public class ReindexProvider { myJobPartitionProvider.getPartitionedUrls(theRequestDetails, urls).forEach(params::addPartitionedUrl); JobInstanceStartRequest request = new JobInstanceStartRequest(); - request.setJobDefinitionId(ReindexAppCtx.JOB_REINDEX); + request.setJobDefinitionId(JOB_REINDEX); request.setParameters(params); Batch2JobStartResponse response = myJobCoordinator.startInstance(theRequestDetails, request); diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexStep.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexStep.java deleted file mode 100644 index 2644f3af89f..00000000000 --- a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexStep.java +++ /dev/null @@ -1,203 +0,0 @@ -/*- - * #%L - * hapi-fhir-storage-batch2-jobs - * %% - * Copyright (C) 2014 - 2024 Smile CDR, Inc. - * %% - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * #L% - */ -package ca.uhn.fhir.batch2.jobs.reindex; - -import ca.uhn.fhir.batch2.api.IJobDataSink; -import ca.uhn.fhir.batch2.api.IJobStepWorker; -import ca.uhn.fhir.batch2.api.JobExecutionFailedException; -import ca.uhn.fhir.batch2.api.RunOutcome; -import ca.uhn.fhir.batch2.api.StepExecutionDetails; -import ca.uhn.fhir.batch2.api.VoidModel; -import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; -import ca.uhn.fhir.jpa.api.dao.DaoRegistry; -import ca.uhn.fhir.jpa.api.dao.IFhirResourceDao; -import ca.uhn.fhir.jpa.api.dao.IFhirSystemDao; -import ca.uhn.fhir.jpa.api.dao.ReindexOutcome; -import ca.uhn.fhir.jpa.api.dao.ReindexParameters; -import ca.uhn.fhir.jpa.api.svc.IIdHelperService; -import ca.uhn.fhir.jpa.dao.tx.HapiTransactionService; -import ca.uhn.fhir.parser.DataFormatException; -import ca.uhn.fhir.rest.api.server.RequestDetails; -import ca.uhn.fhir.rest.api.server.SystemRequestDetails; -import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId; -import ca.uhn.fhir.rest.api.server.storage.TransactionDetails; -import ca.uhn.fhir.rest.server.exceptions.BaseServerResponseException; -import ca.uhn.fhir.util.StopWatch; -import jakarta.annotation.Nonnull; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.transaction.TransactionStatus; -import org.springframework.transaction.support.TransactionCallback; - -import java.util.List; -import java.util.concurrent.TimeUnit; - -public class ReindexStep implements IJobStepWorker { - - public static final int REINDEX_MAX_RETRIES = 10; - - private static final Logger ourLog = LoggerFactory.getLogger(ReindexStep.class); - - @Autowired - private HapiTransactionService myHapiTransactionService; - - @Autowired - private IFhirSystemDao mySystemDao; - - @Autowired - private DaoRegistry myDaoRegistry; - - @Autowired - private IIdHelperService myIdHelperService; - - @Nonnull - @Override - public RunOutcome run( - @Nonnull StepExecutionDetails theStepExecutionDetails, - @Nonnull IJobDataSink theDataSink) - throws JobExecutionFailedException { - - ResourceIdListWorkChunkJson data = theStepExecutionDetails.getData(); - ReindexJobParameters jobParameters = theStepExecutionDetails.getParameters(); - - return doReindex( - data, - theDataSink, - theStepExecutionDetails.getInstance().getInstanceId(), - theStepExecutionDetails.getChunkId(), - jobParameters); - } - - @Nonnull - public RunOutcome doReindex( - ResourceIdListWorkChunkJson data, - IJobDataSink theDataSink, - String theInstanceId, - String theChunkId, - ReindexJobParameters theJobParameters) { - RequestDetails requestDetails = new SystemRequestDetails(); - requestDetails.setRetry(true); - requestDetails.setMaxRetries(REINDEX_MAX_RETRIES); - TransactionDetails transactionDetails = new TransactionDetails(); - ReindexJob reindexJob = new ReindexJob( - data, requestDetails, transactionDetails, theDataSink, theInstanceId, theChunkId, theJobParameters); - - myHapiTransactionService - .withRequest(requestDetails) - .withTransactionDetails(transactionDetails) - .withRequestPartitionId(data.getRequestPartitionId()) - .execute(reindexJob); - - return new RunOutcome(data.size()); - } - - private class ReindexJob implements TransactionCallback { - private final ResourceIdListWorkChunkJson myData; - private final RequestDetails myRequestDetails; - private final TransactionDetails myTransactionDetails; - private final IJobDataSink myDataSink; - private final String myChunkId; - private final String myInstanceId; - private final ReindexJobParameters myJobParameters; - - public ReindexJob( - ResourceIdListWorkChunkJson theData, - RequestDetails theRequestDetails, - TransactionDetails theTransactionDetails, - IJobDataSink theDataSink, - String theInstanceId, - String theChunkId, - ReindexJobParameters theJobParameters) { - myData = theData; - myRequestDetails = theRequestDetails; - myTransactionDetails = theTransactionDetails; - myDataSink = theDataSink; - myInstanceId = theInstanceId; - myChunkId = theChunkId; - myJobParameters = theJobParameters; - myDataSink.setWarningProcessor(new ReindexWarningProcessor()); - } - - @Override - public Void doInTransaction(@Nonnull TransactionStatus theStatus) { - - List persistentIds = myData.getResourcePersistentIds(myIdHelperService); - - ourLog.info( - "Starting reindex work chunk with {} resources - Instance[{}] Chunk[{}]", - persistentIds.size(), - myInstanceId, - myChunkId); - StopWatch sw = new StopWatch(); - - // Prefetch Resources from DB - - boolean reindexSearchParameters = - myJobParameters.getReindexSearchParameters() != ReindexParameters.ReindexSearchParametersEnum.NONE; - mySystemDao.preFetchResources(persistentIds, reindexSearchParameters); - ourLog.info( - "Prefetched {} resources in {} - Instance[{}] Chunk[{}]", - persistentIds.size(), - sw, - myInstanceId, - myChunkId); - - ReindexParameters parameters = new ReindexParameters() - .setReindexSearchParameters(myJobParameters.getReindexSearchParameters()) - .setOptimizeStorage(myJobParameters.getOptimizeStorage()) - .setOptimisticLock(myJobParameters.getOptimisticLock()); - - // Reindex - - sw.restart(); - for (int i = 0; i < myData.size(); i++) { - - String nextResourceType = myData.getResourceType(i); - IFhirResourceDao dao = myDaoRegistry.getResourceDao(nextResourceType); - IResourcePersistentId resourcePersistentId = persistentIds.get(i); - try { - - ReindexOutcome outcome = - dao.reindex(resourcePersistentId, parameters, myRequestDetails, myTransactionDetails); - outcome.getWarnings().forEach(myDataSink::recoveredError); - - } catch (BaseServerResponseException | DataFormatException e) { - String resourceForcedId = myIdHelperService - .translatePidIdToForcedIdWithCache(resourcePersistentId) - .orElse(resourcePersistentId.toString()); - String resourceId = nextResourceType + "/" + resourceForcedId; - ourLog.debug("Failure during reindexing {}", resourceId, e); - myDataSink.recoveredError("Failure reindexing " + resourceId + ": " + e.getMessage()); - } - } - - ourLog.info( - "Finished reindexing {} resources in {} - {}/sec - Instance[{}] Chunk[{}]", - persistentIds.size(), - sw, - sw.formatThroughput(persistentIds.size(), TimeUnit.SECONDS), - myInstanceId, - myChunkId); - - return null; - } - } -} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexUtils.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexUtils.java new file mode 100644 index 00000000000..e54b32a90cc --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexUtils.java @@ -0,0 +1,39 @@ +package ca.uhn.fhir.batch2.jobs.reindex; + +import com.google.common.annotations.VisibleForTesting; + +import java.time.Duration; +import java.time.temporal.ChronoUnit; + +public class ReindexUtils { + + /** + * The reindex job definition id + */ + public static final String JOB_REINDEX = "REINDEX"; + + public static final int REINDEX_MAX_RETRIES = 10; + + private static final Duration RETRY_DELAY = Duration.of(30, ChronoUnit.SECONDS); + + private static Duration myDelay; + + /** + * Returns the retry delay for reindex jobs that require polling. + */ + public static Duration getRetryLaterDelay() { + if (myDelay != null) { + return myDelay; + } + return RETRY_DELAY; + } + + /** + * Sets the retry delay to use for reindex jobs. + * Do not use this in production code! Only test code. + */ + @VisibleForTesting + public static void setRetryDelay(Duration theDuration) { + myDelay = theDuration; + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/models/ReindexResults.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/models/ReindexResults.java new file mode 100644 index 00000000000..0eee14d3ec1 --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/models/ReindexResults.java @@ -0,0 +1,29 @@ +package ca.uhn.fhir.batch2.jobs.reindex.models; + +import ca.uhn.fhir.model.api.IModelJson; +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.util.HashMap; + +public class ReindexResults implements IModelJson { + + /** + * A map of resource type : whether or not the reindex is completed; + * true = more work needed. false (or omitted) = reindex is done + */ + @JsonProperty("resource2NeedsWork") + private HashMap myResourceToHasWorkToComplete; + + public ReindexResults() {} + + public HashMap getResourceToHasWorkToComplete() { + if (myResourceToHasWorkToComplete == null) { + myResourceToHasWorkToComplete = new HashMap<>(); + } + return myResourceToHasWorkToComplete; + } + + public void addResourceTypeToCompletionStatus(String theResourceType, boolean theRequiresMoreWork) { + getResourceToHasWorkToComplete().put(theResourceType, theRequiresMoreWork); + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/svcs/ReindexJobService.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/svcs/ReindexJobService.java new file mode 100644 index 00000000000..39a4273a8b7 --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/svcs/ReindexJobService.java @@ -0,0 +1,38 @@ +package ca.uhn.fhir.batch2.jobs.reindex.svcs; + +import ca.uhn.fhir.jpa.api.dao.DaoRegistry; +import ca.uhn.fhir.jpa.api.dao.IFhirResourceDao; +import ca.uhn.fhir.jpa.api.model.ReindexJobStatus; + +import java.util.Map; + +public class ReindexJobService { + + private final DaoRegistry myDaoRegistry; + + public ReindexJobService(DaoRegistry theRegistry) { + myDaoRegistry = theRegistry; + } + + /** + * Checks if any of the resource types in the map have any pending reindex work waiting. + * This will return true after the first such encounter, and only return false if no + * reindex work is required for any resource. + * @param theResourceTypesToCheckFlag map of resourceType:whether or not to check + * @return true if there's reindex work pending, false otherwise + */ + public boolean anyResourceHasPendingReindexWork(Map theResourceTypesToCheckFlag) { + for (String resourceType : theResourceTypesToCheckFlag.keySet()) { + boolean toCheck = theResourceTypesToCheckFlag.get(resourceType); + if (toCheck) { + IFhirResourceDao dao = myDaoRegistry.getResourceDao(resourceType); + + ReindexJobStatus status = dao.getReindexJobStatus(); + if (status.isHasReindexWorkPending()) { + return true; + } + } + } + return false; + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexJobParametersValidatorV1.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexJobParametersValidatorV1.java new file mode 100644 index 00000000000..a54ef49f9eb --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexJobParametersValidatorV1.java @@ -0,0 +1,57 @@ +/*- + * #%L + * hapi-fhir-storage-batch2-jobs + * %% + * Copyright (C) 2014 - 2024 Smile CDR, Inc. + * %% + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * #L% + */ +package ca.uhn.fhir.batch2.jobs.reindex.v1; + +import ca.uhn.fhir.batch2.api.IJobParametersValidator; +import ca.uhn.fhir.batch2.jobs.parameters.IUrlListValidator; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; +import ca.uhn.fhir.rest.api.server.RequestDetails; +import jakarta.annotation.Nonnull; +import jakarta.annotation.Nullable; + +import java.util.ArrayList; +import java.util.List; + +@Deprecated(forRemoval = true, since = "7.6.0") +public class ReindexJobParametersValidatorV1 implements IJobParametersValidator { + + private final IUrlListValidator myUrlListValidator; + + public ReindexJobParametersValidatorV1(IUrlListValidator theUrlListValidator) { + myUrlListValidator = theUrlListValidator; + } + + @Nullable + @Override + public List validate(RequestDetails theRequestDetails, @Nonnull ReindexJobParameters theParameters) { + List errors = myUrlListValidator.validateUrls(theParameters.getUrls()); + + if (errors == null || errors.isEmpty()) { + // only check if there's no other errors (new list to fix immutable issues) + errors = new ArrayList<>(); + for (String url : theParameters.getUrls()) { + if (url.contains(" ") || url.contains("\n") || url.contains("\t")) { + errors.add("Invalid URL. URL cannot contain spaces : " + url); + } + } + } + return errors; + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexStepV1.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexStepV1.java new file mode 100644 index 00000000000..c9adb93b91c --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexStepV1.java @@ -0,0 +1,117 @@ +/*- + * #%L + * hapi-fhir-storage-batch2-jobs + * %% + * Copyright (C) 2014 - 2024 Smile CDR, Inc. + * %% + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * #L% + */ +package ca.uhn.fhir.batch2.jobs.reindex.v1; + +import ca.uhn.fhir.batch2.api.IJobDataSink; +import ca.uhn.fhir.batch2.api.IJobStepWorker; +import ca.uhn.fhir.batch2.api.JobExecutionFailedException; +import ca.uhn.fhir.batch2.api.RunOutcome; +import ca.uhn.fhir.batch2.api.StepExecutionDetails; +import ca.uhn.fhir.batch2.api.VoidModel; +import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; +import ca.uhn.fhir.batch2.jobs.reindex.models.ReindexResults; +import ca.uhn.fhir.jpa.api.dao.DaoRegistry; +import ca.uhn.fhir.jpa.api.dao.IFhirSystemDao; +import ca.uhn.fhir.jpa.api.svc.IIdHelperService; +import ca.uhn.fhir.jpa.dao.tx.HapiTransactionService; +import ca.uhn.fhir.rest.api.server.RequestDetails; +import ca.uhn.fhir.rest.api.server.SystemRequestDetails; +import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId; +import ca.uhn.fhir.rest.api.server.storage.TransactionDetails; +import jakarta.annotation.Nonnull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.REINDEX_MAX_RETRIES; + +@Deprecated(forRemoval = true, since = "7.6.0") +public class ReindexStepV1 implements IJobStepWorker { + + private static final Logger ourLog = LoggerFactory.getLogger(ReindexStepV1.class); + + private final HapiTransactionService myHapiTransactionService; + + private final IFhirSystemDao mySystemDao; + + private final DaoRegistry myDaoRegistry; + + private final IIdHelperService> myIdHelperService; + + public ReindexStepV1( + HapiTransactionService theHapiTransactionService, + IFhirSystemDao theSystemDao, + DaoRegistry theRegistry, + IIdHelperService> theIdHelperService) { + myDaoRegistry = theRegistry; + myHapiTransactionService = theHapiTransactionService; + mySystemDao = theSystemDao; + myIdHelperService = theIdHelperService; + } + + @Nonnull + @Override + public RunOutcome run( + @Nonnull StepExecutionDetails theStepExecutionDetails, + @Nonnull IJobDataSink theDataSink) + throws JobExecutionFailedException { + + ResourceIdListWorkChunkJson data = theStepExecutionDetails.getData(); + ReindexJobParameters jobParameters = theStepExecutionDetails.getParameters(); + + doReindex( + data, + theDataSink, + theStepExecutionDetails.getInstance().getInstanceId(), + theStepExecutionDetails.getChunkId(), + jobParameters); + + return new RunOutcome(data.size()); + } + + public ReindexResults doReindex( + ResourceIdListWorkChunkJson data, + IJobDataSink theDataSink, + String theInstanceId, + String theChunkId, + ReindexJobParameters theJobParameters) { + RequestDetails requestDetails = new SystemRequestDetails(); + requestDetails.setRetry(true); + requestDetails.setMaxRetries(REINDEX_MAX_RETRIES); + + TransactionDetails transactionDetails = new TransactionDetails(); + ReindexTaskV1.JobParameters jp = new ReindexTaskV1.JobParameters(); + jp.setData(data) + .setRequestDetails(requestDetails) + .setTransactionDetails(transactionDetails) + .setDataSink(theDataSink) + .setInstanceId(theInstanceId) + .setChunkId(theChunkId) + .setJobParameters(theJobParameters); + + ReindexTaskV1 reindexJob = new ReindexTaskV1(jp, myDaoRegistry, mySystemDao, myIdHelperService); + + return myHapiTransactionService + .withRequest(requestDetails) + .withTransactionDetails(transactionDetails) + .withRequestPartitionId(data.getRequestPartitionId()) + .execute(reindexJob); + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexTaskV1.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexTaskV1.java new file mode 100644 index 00000000000..f6dd6db4a8c --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexTaskV1.java @@ -0,0 +1,202 @@ +package ca.uhn.fhir.batch2.jobs.reindex.v1; + +import ca.uhn.fhir.batch2.api.IJobDataSink; +import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexWarningProcessor; +import ca.uhn.fhir.batch2.jobs.reindex.models.ReindexResults; +import ca.uhn.fhir.jpa.api.dao.DaoRegistry; +import ca.uhn.fhir.jpa.api.dao.IFhirResourceDao; +import ca.uhn.fhir.jpa.api.dao.IFhirSystemDao; +import ca.uhn.fhir.jpa.api.dao.ReindexOutcome; +import ca.uhn.fhir.jpa.api.dao.ReindexParameters; +import ca.uhn.fhir.jpa.api.svc.IIdHelperService; +import ca.uhn.fhir.parser.DataFormatException; +import ca.uhn.fhir.rest.api.server.RequestDetails; +import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId; +import ca.uhn.fhir.rest.api.server.storage.TransactionDetails; +import ca.uhn.fhir.rest.server.exceptions.BaseServerResponseException; +import ca.uhn.fhir.util.StopWatch; +import jakarta.annotation.Nonnull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionCallback; + +import java.util.List; +import java.util.concurrent.TimeUnit; + +@Deprecated(forRemoval = true, since = "7.6.0") +public class ReindexTaskV1 implements TransactionCallback { + private static final Logger ourLog = LoggerFactory.getLogger(ReindexTaskV1.class); + + public static class JobParameters { + private ResourceIdListWorkChunkJson myData; + private RequestDetails myRequestDetails; + private TransactionDetails myTransactionDetails; + private IJobDataSink myDataSink; + private String myChunkId; + private String myInstanceId; + private ReindexJobParameters myJobParameters; + + public ResourceIdListWorkChunkJson getData() { + return myData; + } + + public JobParameters setData(ResourceIdListWorkChunkJson theData) { + myData = theData; + return this; + } + + public RequestDetails getRequestDetails() { + return myRequestDetails; + } + + public JobParameters setRequestDetails(RequestDetails theRequestDetails) { + myRequestDetails = theRequestDetails; + return this; + } + + public TransactionDetails getTransactionDetails() { + return myTransactionDetails; + } + + public JobParameters setTransactionDetails(TransactionDetails theTransactionDetails) { + myTransactionDetails = theTransactionDetails; + return this; + } + + public IJobDataSink getDataSink() { + return myDataSink; + } + + public JobParameters setDataSink(IJobDataSink theDataSink) { + myDataSink = theDataSink; + return this; + } + + public String getChunkId() { + return myChunkId; + } + + public JobParameters setChunkId(String theChunkId) { + myChunkId = theChunkId; + return this; + } + + public String getInstanceId() { + return myInstanceId; + } + + public JobParameters setInstanceId(String theInstanceId) { + myInstanceId = theInstanceId; + return this; + } + + public ReindexJobParameters getJobParameters() { + return myJobParameters; + } + + public JobParameters setJobParameters(ReindexJobParameters theJobParameters) { + myJobParameters = theJobParameters; + return this; + } + } + + private final DaoRegistry myDaoRegistry; + private final IFhirSystemDao mySystemDao; + + private final IIdHelperService> myIdHelperService; + + private final ResourceIdListWorkChunkJson myData; + private final RequestDetails myRequestDetails; + private final TransactionDetails myTransactionDetails; + private final IJobDataSink myDataSink; + private final String myChunkId; + private final String myInstanceId; + private final ReindexJobParameters myJobParameters; + + public ReindexTaskV1( + JobParameters theJobParameters, + DaoRegistry theRegistry, + IFhirSystemDao theSystemDao, + IIdHelperService> theIdHelperService) { + myDaoRegistry = theRegistry; + mySystemDao = theSystemDao; + myIdHelperService = theIdHelperService; + + myData = theJobParameters.getData(); + myRequestDetails = theJobParameters.getRequestDetails(); + myTransactionDetails = theJobParameters.getTransactionDetails(); + myDataSink = theJobParameters.getDataSink(); + myInstanceId = theJobParameters.getInstanceId(); + myChunkId = theJobParameters.getChunkId(); + myJobParameters = theJobParameters.getJobParameters(); + myDataSink.setWarningProcessor(new ReindexWarningProcessor()); + } + + @Override + public ReindexResults doInTransaction(@Nonnull TransactionStatus theStatus) { + List> persistentIds = myData.getResourcePersistentIds(myIdHelperService); + + ourLog.info( + "Starting reindex work chunk with {} resources - Instance[{}] Chunk[{}]", + persistentIds.size(), + myInstanceId, + myChunkId); + StopWatch sw = new StopWatch(); + ReindexResults reindexResults = new ReindexResults(); + + // Prefetch Resources from DB + boolean reindexSearchParameters = + myJobParameters.getReindexSearchParameters() != ReindexParameters.ReindexSearchParametersEnum.NONE; + mySystemDao.preFetchResources(persistentIds, reindexSearchParameters); + ourLog.info( + "Prefetched {} resources in {} - Instance[{}] Chunk[{}]", + persistentIds.size(), + sw, + myInstanceId, + myChunkId); + + ReindexParameters parameters = new ReindexParameters() + .setReindexSearchParameters(myJobParameters.getReindexSearchParameters()) + .setOptimizeStorage(myJobParameters.getOptimizeStorage()) + .setOptimisticLock(myJobParameters.getOptimisticLock()); + + // Reindex + + sw.restart(); + for (int i = 0; i < myData.size(); i++) { + + String nextResourceType = myData.getResourceType(i); + IFhirResourceDao dao = myDaoRegistry.getResourceDao(nextResourceType); + IResourcePersistentId resourcePersistentId = persistentIds.get(i); + try { + + ReindexOutcome outcome = + dao.reindex(resourcePersistentId, parameters, myRequestDetails, myTransactionDetails); + + outcome.getWarnings().forEach(myDataSink::recoveredError); + reindexResults.addResourceTypeToCompletionStatus(nextResourceType, outcome.isHasPendingWork()); + + } catch (BaseServerResponseException | DataFormatException e) { + String resourceForcedId = myIdHelperService + .translatePidIdToForcedIdWithCache(resourcePersistentId) + .orElse(resourcePersistentId.toString()); + String resourceId = nextResourceType + "/" + resourceForcedId; + ourLog.error("Failure during reindexing {}", resourceId, e); + myDataSink.recoveredError("Failure reindexing " + resourceId + ": " + e.getMessage()); + } + } + + ourLog.info( + "Finished reindexing {} resources in {} - {}/sec - Instance[{}] Chunk[{}]", + persistentIds.size(), + sw, + sw.formatThroughput(persistentIds.size(), TimeUnit.SECONDS), + myInstanceId, + myChunkId); + + return reindexResults; + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexV1Config.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexV1Config.java new file mode 100644 index 00000000000..8bda0695a5c --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v1/ReindexV1Config.java @@ -0,0 +1,101 @@ +package ca.uhn.fhir.batch2.jobs.reindex.v1; + +import ca.uhn.fhir.batch2.api.IJobStepWorker; +import ca.uhn.fhir.batch2.api.VoidModel; +import ca.uhn.fhir.batch2.jobs.chunk.ChunkRangeJson; +import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; +import ca.uhn.fhir.batch2.jobs.parameters.UrlListValidator; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; +import ca.uhn.fhir.batch2.jobs.reindex.svcs.ReindexJobService; +import ca.uhn.fhir.batch2.jobs.step.GenerateRangeChunksStep; +import ca.uhn.fhir.batch2.jobs.step.LoadIdsStep; +import ca.uhn.fhir.batch2.model.JobDefinition; +import ca.uhn.fhir.jpa.api.dao.DaoRegistry; +import ca.uhn.fhir.jpa.api.dao.IFhirSystemDao; +import ca.uhn.fhir.jpa.api.svc.IBatch2DaoSvc; +import ca.uhn.fhir.jpa.api.svc.IIdHelperService; +import ca.uhn.fhir.jpa.dao.tx.HapiTransactionService; +import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId; +import ca.uhn.fhir.rest.server.provider.ProviderConstants; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; + +@Deprecated(forRemoval = true, since = "7.6.0") +@Configuration +public class ReindexV1Config { + @Autowired + private ReindexJobService myReindexJobService; + + @Autowired + private HapiTransactionService myHapiTransactionService; + + @Autowired + private IFhirSystemDao mySystemDao; + + @Autowired + private DaoRegistry myRegistry; + + @Autowired + private IIdHelperService> myIdHelperService; + + @Autowired + @Qualifier("reindexGenerateRangeChunkStepV1") + private IJobStepWorker myReindexGenerateRangeChunkStep; + + @Autowired + @Qualifier("reindexLoadIdsStepV1") + private IJobStepWorker myReindexLoadIdsStep; + + @Autowired + private ReindexJobParametersValidatorV1 myReindexJobParametersValidatorV1; + + // Version 1 + @Bean + public JobDefinition reindexJobDefinitionV1() { + return JobDefinition.newBuilder() + .setJobDefinitionId(JOB_REINDEX) + .setJobDescription("Reindex resources") + .setJobDefinitionVersion(1) + .setParametersType(ReindexJobParameters.class) + .setParametersValidator(myReindexJobParametersValidatorV1) + .gatedExecution() + .addFirstStep( + "generate-ranges", + "Generate data ranges to reindex", + ChunkRangeJson.class, + myReindexGenerateRangeChunkStep) + .addIntermediateStep( + "load-ids", + "Load IDs of resources to reindex", + ResourceIdListWorkChunkJson.class, + myReindexLoadIdsStep) + .addLastStep("reindex-start", "Start the resource reindex", reindexStepV1()) + .build(); + } + + @Bean + public ReindexStepV1 reindexStepV1() { + return new ReindexStepV1(myHapiTransactionService, mySystemDao, myRegistry, myIdHelperService); + } + + @Bean("reindexGenerateRangeChunkStepV1") + public IJobStepWorker reindexGenerateRangeChunksStep() { + return new GenerateRangeChunksStep<>(); + } + + @Bean("reindexLoadIdsStepV1") + public IJobStepWorker reindexLoadIdsStep( + IBatch2DaoSvc theBatch2DaoSvc) { + return new LoadIdsStep<>(theBatch2DaoSvc); + } + + @Bean + public ReindexJobParametersValidatorV1 reindexJobParametersValidatorV1(IBatch2DaoSvc theBatch2DaoSvc) { + return new ReindexJobParametersValidatorV1( + new UrlListValidator(ProviderConstants.OPERATION_REINDEX, theBatch2DaoSvc)); + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/CheckPendingReindexWorkStep.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/CheckPendingReindexWorkStep.java new file mode 100644 index 00000000000..9a9dfd294f9 --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/CheckPendingReindexWorkStep.java @@ -0,0 +1,42 @@ +package ca.uhn.fhir.batch2.jobs.reindex.v2; + +import ca.uhn.fhir.batch2.api.IJobDataSink; +import ca.uhn.fhir.batch2.api.IJobStepWorker; +import ca.uhn.fhir.batch2.api.JobExecutionFailedException; +import ca.uhn.fhir.batch2.api.RetryChunkLaterException; +import ca.uhn.fhir.batch2.api.RunOutcome; +import ca.uhn.fhir.batch2.api.StepExecutionDetails; +import ca.uhn.fhir.batch2.api.VoidModel; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils; +import ca.uhn.fhir.batch2.jobs.reindex.models.ReindexResults; +import ca.uhn.fhir.batch2.jobs.reindex.svcs.ReindexJobService; +import ca.uhn.fhir.i18n.Msg; +import jakarta.annotation.Nonnull; + +public class CheckPendingReindexWorkStep implements IJobStepWorker { + + private final ReindexJobService myReindexJobService; + + public CheckPendingReindexWorkStep(ReindexJobService theReindexJobService) { + myReindexJobService = theReindexJobService; + } + + @Nonnull + @Override + public RunOutcome run( + @Nonnull StepExecutionDetails theStepExecutionDetails, + @Nonnull IJobDataSink theDataSink) + throws JobExecutionFailedException { + + ReindexResults results = theStepExecutionDetails.getData(); + + if (!results.getResourceToHasWorkToComplete().isEmpty()) { + if (myReindexJobService.anyResourceHasPendingReindexWork(results.getResourceToHasWorkToComplete())) { + throw new RetryChunkLaterException(Msg.code(2553), ReindexUtils.getRetryLaterDelay()); + } + } + + return RunOutcome.SUCCESS; + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexJobParametersValidator.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexJobParametersValidatorV2.java similarity index 85% rename from hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexJobParametersValidator.java rename to hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexJobParametersValidatorV2.java index b7560a8ef57..9bfbbfe6564 100644 --- a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexJobParametersValidator.java +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexJobParametersValidatorV2.java @@ -17,10 +17,11 @@ * limitations under the License. * #L% */ -package ca.uhn.fhir.batch2.jobs.reindex; +package ca.uhn.fhir.batch2.jobs.reindex.v2; import ca.uhn.fhir.batch2.api.IJobParametersValidator; import ca.uhn.fhir.batch2.jobs.parameters.IUrlListValidator; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; import ca.uhn.fhir.rest.api.server.RequestDetails; import jakarta.annotation.Nonnull; import jakarta.annotation.Nullable; @@ -28,11 +29,11 @@ import jakarta.annotation.Nullable; import java.util.ArrayList; import java.util.List; -public class ReindexJobParametersValidator implements IJobParametersValidator { +public class ReindexJobParametersValidatorV2 implements IJobParametersValidator { private final IUrlListValidator myUrlListValidator; - public ReindexJobParametersValidator(IUrlListValidator theUrlListValidator) { + public ReindexJobParametersValidatorV2(IUrlListValidator theUrlListValidator) { myUrlListValidator = theUrlListValidator; } diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexStepV2.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexStepV2.java new file mode 100644 index 00000000000..adfad38c1c1 --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexStepV2.java @@ -0,0 +1,118 @@ +package ca.uhn.fhir.batch2.jobs.reindex.v2; + +import ca.uhn.fhir.batch2.api.IJobDataSink; +import ca.uhn.fhir.batch2.api.IJobStepWorker; +import ca.uhn.fhir.batch2.api.JobExecutionFailedException; +import ca.uhn.fhir.batch2.api.RetryChunkLaterException; +import ca.uhn.fhir.batch2.api.RunOutcome; +import ca.uhn.fhir.batch2.api.StepExecutionDetails; +import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils; +import ca.uhn.fhir.batch2.jobs.reindex.models.ReindexResults; +import ca.uhn.fhir.batch2.jobs.reindex.svcs.ReindexJobService; +import ca.uhn.fhir.i18n.Msg; +import ca.uhn.fhir.jpa.api.dao.DaoRegistry; +import ca.uhn.fhir.jpa.api.dao.IFhirSystemDao; +import ca.uhn.fhir.jpa.api.svc.IIdHelperService; +import ca.uhn.fhir.jpa.dao.tx.HapiTransactionService; +import ca.uhn.fhir.rest.api.server.RequestDetails; +import ca.uhn.fhir.rest.api.server.SystemRequestDetails; +import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId; +import ca.uhn.fhir.rest.api.server.storage.TransactionDetails; +import jakarta.annotation.Nonnull; + +import java.util.HashMap; +import java.util.Map; + +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.REINDEX_MAX_RETRIES; + +public class ReindexStepV2 + implements IJobStepWorker { + + private final ReindexJobService myReindexJobService; + private final HapiTransactionService myHapiTransactionService; + + private final IFhirSystemDao mySystemDao; + + private final DaoRegistry myDaoRegistry; + + private final IIdHelperService> myIdHelperService; + + public ReindexStepV2( + ReindexJobService theJobService, + HapiTransactionService theHapiTransactionService, + IFhirSystemDao theSystemDao, + DaoRegistry theRegistry, + IIdHelperService> theIdHelperService) { + myDaoRegistry = theRegistry; + myHapiTransactionService = theHapiTransactionService; + mySystemDao = theSystemDao; + myIdHelperService = theIdHelperService; + myReindexJobService = theJobService; + } + + @Nonnull + @Override + public RunOutcome run( + @Nonnull StepExecutionDetails theStepExecutionDetails, + @Nonnull IJobDataSink theDataSink) + throws JobExecutionFailedException { + ResourceIdListWorkChunkJson data = theStepExecutionDetails.getData(); + ReindexJobParameters jobParameters = theStepExecutionDetails.getParameters(); + + // This is not strictly necessary; + // but we'll ensure that no outstanding "reindex work" + // is waiting to be completed, so that when we do + // our reindex work here, it won't skip over that data + Map resourceTypesToCheckFlag = new HashMap<>(); + data.getTypedPids().forEach(id -> { + // we don't really care about duplicates; we check by resource type + resourceTypesToCheckFlag.put(id.getResourceType(), true); + }); + if (myReindexJobService.anyResourceHasPendingReindexWork(resourceTypesToCheckFlag)) { + + throw new RetryChunkLaterException(Msg.code(2552), ReindexUtils.getRetryLaterDelay()); + } + + ReindexResults results = doReindex( + data, + theDataSink, + theStepExecutionDetails.getInstance().getInstanceId(), + theStepExecutionDetails.getChunkId(), + jobParameters); + + theDataSink.accept(results); + + return new RunOutcome(data.size()); + } + + public ReindexResults doReindex( + ResourceIdListWorkChunkJson data, + IJobDataSink theDataSink, + String theInstanceId, + String theChunkId, + ReindexJobParameters theJobParameters) { + RequestDetails requestDetails = new SystemRequestDetails(); + requestDetails.setRetry(true); + requestDetails.setMaxRetries(REINDEX_MAX_RETRIES); + + TransactionDetails transactionDetails = new TransactionDetails(); + ReindexTaskV2.JobParameters jp = new ReindexTaskV2.JobParameters(); + jp.setData(data) + .setRequestDetails(requestDetails) + .setTransactionDetails(transactionDetails) + .setDataSink(theDataSink) + .setInstanceId(theInstanceId) + .setChunkId(theChunkId) + .setJobParameters(theJobParameters); + + ReindexTaskV2 reindexJob = new ReindexTaskV2(jp, myDaoRegistry, mySystemDao, myIdHelperService); + + return myHapiTransactionService + .withRequest(requestDetails) + .withTransactionDetails(transactionDetails) + .withRequestPartitionId(data.getRequestPartitionId()) + .execute(reindexJob); + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexTaskV2.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexTaskV2.java new file mode 100644 index 00000000000..0e77b74f6e6 --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexTaskV2.java @@ -0,0 +1,201 @@ +package ca.uhn.fhir.batch2.jobs.reindex.v2; + +import ca.uhn.fhir.batch2.api.IJobDataSink; +import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexWarningProcessor; +import ca.uhn.fhir.batch2.jobs.reindex.models.ReindexResults; +import ca.uhn.fhir.jpa.api.dao.DaoRegistry; +import ca.uhn.fhir.jpa.api.dao.IFhirResourceDao; +import ca.uhn.fhir.jpa.api.dao.IFhirSystemDao; +import ca.uhn.fhir.jpa.api.dao.ReindexOutcome; +import ca.uhn.fhir.jpa.api.dao.ReindexParameters; +import ca.uhn.fhir.jpa.api.svc.IIdHelperService; +import ca.uhn.fhir.parser.DataFormatException; +import ca.uhn.fhir.rest.api.server.RequestDetails; +import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId; +import ca.uhn.fhir.rest.api.server.storage.TransactionDetails; +import ca.uhn.fhir.rest.server.exceptions.BaseServerResponseException; +import ca.uhn.fhir.util.StopWatch; +import jakarta.annotation.Nonnull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionCallback; + +import java.util.List; +import java.util.concurrent.TimeUnit; + +public class ReindexTaskV2 implements TransactionCallback { + private static final Logger ourLog = LoggerFactory.getLogger(ReindexTaskV2.class); + + public static class JobParameters { + private ResourceIdListWorkChunkJson myData; + private RequestDetails myRequestDetails; + private TransactionDetails myTransactionDetails; + private IJobDataSink myDataSink; + private String myChunkId; + private String myInstanceId; + private ReindexJobParameters myJobParameters; + + public ResourceIdListWorkChunkJson getData() { + return myData; + } + + public JobParameters setData(ResourceIdListWorkChunkJson theData) { + myData = theData; + return this; + } + + public RequestDetails getRequestDetails() { + return myRequestDetails; + } + + public JobParameters setRequestDetails(RequestDetails theRequestDetails) { + myRequestDetails = theRequestDetails; + return this; + } + + public TransactionDetails getTransactionDetails() { + return myTransactionDetails; + } + + public JobParameters setTransactionDetails(TransactionDetails theTransactionDetails) { + myTransactionDetails = theTransactionDetails; + return this; + } + + public IJobDataSink getDataSink() { + return myDataSink; + } + + public JobParameters setDataSink(IJobDataSink theDataSink) { + myDataSink = theDataSink; + return this; + } + + public String getChunkId() { + return myChunkId; + } + + public JobParameters setChunkId(String theChunkId) { + myChunkId = theChunkId; + return this; + } + + public String getInstanceId() { + return myInstanceId; + } + + public JobParameters setInstanceId(String theInstanceId) { + myInstanceId = theInstanceId; + return this; + } + + public ReindexJobParameters getJobParameters() { + return myJobParameters; + } + + public JobParameters setJobParameters(ReindexJobParameters theJobParameters) { + myJobParameters = theJobParameters; + return this; + } + } + + private final DaoRegistry myDaoRegistry; + private final IFhirSystemDao mySystemDao; + + private final IIdHelperService> myIdHelperService; + + private final ResourceIdListWorkChunkJson myData; + private final RequestDetails myRequestDetails; + private final TransactionDetails myTransactionDetails; + private final IJobDataSink myDataSink; + private final String myChunkId; + private final String myInstanceId; + private final ReindexJobParameters myJobParameters; + + public ReindexTaskV2( + JobParameters theJobParameters, + DaoRegistry theRegistry, + IFhirSystemDao theSystemDao, + IIdHelperService> theIdHelperService) { + myDaoRegistry = theRegistry; + mySystemDao = theSystemDao; + myIdHelperService = theIdHelperService; + + myData = theJobParameters.getData(); + myRequestDetails = theJobParameters.getRequestDetails(); + myTransactionDetails = theJobParameters.getTransactionDetails(); + myDataSink = theJobParameters.getDataSink(); + myInstanceId = theJobParameters.getInstanceId(); + myChunkId = theJobParameters.getChunkId(); + myJobParameters = theJobParameters.getJobParameters(); + myDataSink.setWarningProcessor(new ReindexWarningProcessor()); + } + + @Override + public ReindexResults doInTransaction(@Nonnull TransactionStatus theStatus) { + List> persistentIds = myData.getResourcePersistentIds(myIdHelperService); + + ourLog.info( + "Starting reindex work chunk with {} resources - Instance[{}] Chunk[{}]", + persistentIds.size(), + myInstanceId, + myChunkId); + StopWatch sw = new StopWatch(); + ReindexResults reindexResults = new ReindexResults(); + + // Prefetch Resources from DB + boolean reindexSearchParameters = + myJobParameters.getReindexSearchParameters() != ReindexParameters.ReindexSearchParametersEnum.NONE; + mySystemDao.preFetchResources(persistentIds, reindexSearchParameters); + ourLog.info( + "Prefetched {} resources in {} - Instance[{}] Chunk[{}]", + persistentIds.size(), + sw, + myInstanceId, + myChunkId); + + ReindexParameters parameters = new ReindexParameters() + .setReindexSearchParameters(myJobParameters.getReindexSearchParameters()) + .setOptimizeStorage(myJobParameters.getOptimizeStorage()) + .setOptimisticLock(myJobParameters.getOptimisticLock()); + + // Reindex + + sw.restart(); + for (int i = 0; i < myData.size(); i++) { + + String nextResourceType = myData.getResourceType(i); + IFhirResourceDao dao = myDaoRegistry.getResourceDao(nextResourceType); + IResourcePersistentId resourcePersistentId = persistentIds.get(i); + try { + + ReindexOutcome outcome = + dao.reindex(resourcePersistentId, parameters, myRequestDetails, myTransactionDetails); + + outcome.getWarnings().forEach(myDataSink::recoveredError); + reindexResults.addResourceTypeToCompletionStatus(nextResourceType, outcome.isHasPendingWork()); + + } catch (BaseServerResponseException | DataFormatException e) { + String resourceForcedId = myIdHelperService + .translatePidIdToForcedIdWithCache(resourcePersistentId) + .orElse(resourcePersistentId.toString()); + String resourceId = nextResourceType + "/" + resourceForcedId; + ourLog.error("Failure during reindexing {}", resourceId, e); + myDataSink.recoveredError("Failure reindexing " + resourceId + ": " + e.getMessage()); + } + } + + ourLog.info( + "Finished reindexing {} resources in {} - {}/sec - Instance[{}] Chunk[{}]", + persistentIds.size(), + sw, + sw.formatThroughput(persistentIds.size(), TimeUnit.SECONDS), + myInstanceId, + myChunkId); + + return reindexResults; + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexV2Config.java b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexV2Config.java new file mode 100644 index 00000000000..6cc4553c64b --- /dev/null +++ b/hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/v2/ReindexV2Config.java @@ -0,0 +1,110 @@ +package ca.uhn.fhir.batch2.jobs.reindex.v2; + +import ca.uhn.fhir.batch2.api.IJobStepWorker; +import ca.uhn.fhir.batch2.api.VoidModel; +import ca.uhn.fhir.batch2.jobs.chunk.ChunkRangeJson; +import ca.uhn.fhir.batch2.jobs.chunk.ResourceIdListWorkChunkJson; +import ca.uhn.fhir.batch2.jobs.parameters.UrlListValidator; +import ca.uhn.fhir.batch2.jobs.reindex.ReindexJobParameters; +import ca.uhn.fhir.batch2.jobs.reindex.models.ReindexResults; +import ca.uhn.fhir.batch2.jobs.reindex.svcs.ReindexJobService; +import ca.uhn.fhir.batch2.jobs.step.GenerateRangeChunksStep; +import ca.uhn.fhir.batch2.jobs.step.LoadIdsStep; +import ca.uhn.fhir.batch2.model.JobDefinition; +import ca.uhn.fhir.jpa.api.dao.DaoRegistry; +import ca.uhn.fhir.jpa.api.dao.IFhirSystemDao; +import ca.uhn.fhir.jpa.api.svc.IBatch2DaoSvc; +import ca.uhn.fhir.jpa.api.svc.IIdHelperService; +import ca.uhn.fhir.jpa.dao.tx.HapiTransactionService; +import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId; +import ca.uhn.fhir.rest.server.provider.ProviderConstants; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import static ca.uhn.fhir.batch2.jobs.reindex.ReindexUtils.JOB_REINDEX; + +@Configuration +public class ReindexV2Config { + + @Autowired + private ReindexJobService myReindexJobService; + + @Autowired + private HapiTransactionService myHapiTransactionService; + + @Autowired + private IFhirSystemDao mySystemDao; + + @Autowired + private DaoRegistry myRegistry; + + @Autowired + private IIdHelperService> myIdHelperService; + + @Autowired + @Qualifier("reindexGenerateRangeChunkStepV2") + private IJobStepWorker myReindexGenerateRangeChunkStep; + + @Autowired + @Qualifier("reindexLoadIdsStepV2") + private IJobStepWorker myReindexLoadIdsStep; + + @Autowired + private ReindexJobParametersValidatorV2 myReindexJobParametersValidator; + + // Version 2 + @Bean + public JobDefinition reindexJobDefinitionV2() { + return JobDefinition.newBuilder() + .setJobDefinitionId(JOB_REINDEX) + .setJobDescription("Reindex resources") + .setJobDefinitionVersion(2) + .setParametersType(ReindexJobParameters.class) + .setParametersValidator(myReindexJobParametersValidator) + .gatedExecution() + .addFirstStep( + "generate-ranges", + "Generate data ranges to reindex", + ChunkRangeJson.class, + myReindexGenerateRangeChunkStep) + .addIntermediateStep( + "load-ids", + "Load IDs of resources to reindex", + ResourceIdListWorkChunkJson.class, + myReindexLoadIdsStep) + .addIntermediateStep( + "reindex-start", "Perform the resource reindex", ReindexResults.class, reindexStepV2()) + .addLastStep("reindex-pending-work", "Waits for reindex work to complete.", pendingWorkStep()) + .build(); + } + + @Bean + public CheckPendingReindexWorkStep pendingWorkStep() { + return new CheckPendingReindexWorkStep(myReindexJobService); + } + + @Bean + public ReindexStepV2 reindexStepV2() { + return new ReindexStepV2( + myReindexJobService, myHapiTransactionService, mySystemDao, myRegistry, myIdHelperService); + } + + @Bean("reindexGenerateRangeChunkStepV2") + public IJobStepWorker reindexGenerateRangeChunksStep() { + return new GenerateRangeChunksStep<>(); + } + + @Bean("reindexLoadIdsStepV2") + public IJobStepWorker reindexLoadIdsStep( + IBatch2DaoSvc theBatch2DaoSvc) { + return new LoadIdsStep<>(theBatch2DaoSvc); + } + + @Bean + public ReindexJobParametersValidatorV2 reindexJobParametersValidatorV2(IBatch2DaoSvc theBatch2DaoSvc) { + return new ReindexJobParametersValidatorV2( + new UrlListValidator(ProviderConstants.OPERATION_REINDEX, theBatch2DaoSvc)); + } +} diff --git a/hapi-fhir-storage-batch2-jobs/src/test/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexJobParametersValidatorTest.java b/hapi-fhir-storage-batch2-jobs/src/test/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexTaskParametersValidatorTest.java similarity index 86% rename from hapi-fhir-storage-batch2-jobs/src/test/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexJobParametersValidatorTest.java rename to hapi-fhir-storage-batch2-jobs/src/test/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexTaskParametersValidatorTest.java index 0fe1c00e461..f03db306889 100644 --- a/hapi-fhir-storage-batch2-jobs/src/test/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexJobParametersValidatorTest.java +++ b/hapi-fhir-storage-batch2-jobs/src/test/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexTaskParametersValidatorTest.java @@ -1,6 +1,7 @@ package ca.uhn.fhir.batch2.jobs.reindex; import ca.uhn.fhir.batch2.jobs.parameters.UrlListValidator; +import ca.uhn.fhir.batch2.jobs.reindex.v1.ReindexJobParametersValidatorV1; import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -13,13 +14,13 @@ import java.util.List; import static org.assertj.core.api.Assertions.assertThat; @ExtendWith(MockitoExtension.class) -public class ReindexJobParametersValidatorTest { +public class ReindexTaskParametersValidatorTest { @Mock private UrlListValidator myListValidator; @InjectMocks - private ReindexJobParametersValidator myValidator; + private ReindexJobParametersValidatorV1 myValidator; @ParameterizedTest @ValueSource(strings = { "\n", " ", "\t" }) diff --git a/hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/api/RetryChunkLaterException.java b/hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/api/RetryChunkLaterException.java index ec35cb770d2..0d132fc7191 100644 --- a/hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/api/RetryChunkLaterException.java +++ b/hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/api/RetryChunkLaterException.java @@ -38,11 +38,18 @@ public class RetryChunkLaterException extends RuntimeException { private final Duration myNextPollDuration; public RetryChunkLaterException() { - this(ONE_MINUTE); + this("", ONE_MINUTE); } + /** + * For HAPI exceptions, use {@link RetryChunkLaterException#RetryChunkLaterException(String, Duration)} + */ public RetryChunkLaterException(Duration theDuration) { - super(); + this("", theDuration); + } + + public RetryChunkLaterException(String theCode, Duration theDuration) { + super(theCode); this.myNextPollDuration = theDuration; } diff --git a/hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/jobs/chunk/ResourceIdListWorkChunkJson.java b/hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/jobs/chunk/ResourceIdListWorkChunkJson.java index b7969f1ae1c..0ecacb9bdf6 100644 --- a/hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/jobs/chunk/ResourceIdListWorkChunkJson.java +++ b/hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/jobs/chunk/ResourceIdListWorkChunkJson.java @@ -62,7 +62,7 @@ public class ResourceIdListWorkChunkJson implements IModelJson { return myRequestPartitionId; } - private List getTypedPids() { + public List getTypedPids() { if (myTypedPids == null) { myTypedPids = new ArrayList<>(); } diff --git a/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/dao/IFhirResourceDao.java b/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/dao/IFhirResourceDao.java index 7082fed8240..d918f5572ce 100644 --- a/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/dao/IFhirResourceDao.java +++ b/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/dao/IFhirResourceDao.java @@ -26,6 +26,7 @@ import ca.uhn.fhir.jpa.api.model.DeleteConflictList; import ca.uhn.fhir.jpa.api.model.DeleteMethodOutcome; import ca.uhn.fhir.jpa.api.model.ExpungeOptions; import ca.uhn.fhir.jpa.api.model.ExpungeOutcome; +import ca.uhn.fhir.jpa.api.model.ReindexJobStatus; import ca.uhn.fhir.jpa.model.cross.IBasePersistedResource; import ca.uhn.fhir.jpa.model.entity.TagTypeEnum; import ca.uhn.fhir.jpa.searchparam.SearchParameterMap; @@ -315,12 +316,24 @@ public interface IFhirResourceDao extends IDao { * @param theResourcePersistentId The ID * @return */ + @SuppressWarnings("rawtypes") ReindexOutcome reindex( IResourcePersistentId theResourcePersistentId, ReindexParameters theReindexParameters, RequestDetails theRequest, TransactionDetails theTransactionDetails); + /** + * Returns ReindexJobStatus information object that tells the caller + * if a reindex job is still in progress or not. + * + * If the implementing DAO requires additional work during reindexing, + * this is the method to override. + */ + default ReindexJobStatus getReindexJobStatus() { + return ReindexJobStatus.NO_WORK_NEEDED; + } + void removeTag( IIdType theId, TagTypeEnum theTagType, String theSystem, String theCode, RequestDetails theRequestDetails); diff --git a/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/dao/ReindexOutcome.java b/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/dao/ReindexOutcome.java index 8c7eb7fa5b9..744e1def01f 100644 --- a/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/dao/ReindexOutcome.java +++ b/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/dao/ReindexOutcome.java @@ -29,6 +29,11 @@ public class ReindexOutcome { private List myWarnings; + /** + * True if there is additional (async) work to wait on. + */ + private boolean myHasPendingWork; + public List getWarnings() { return defaultIfNull(myWarnings, Collections.emptyList()); } @@ -39,4 +44,12 @@ public class ReindexOutcome { } myWarnings.add(theWarning); } + + public boolean isHasPendingWork() { + return myHasPendingWork; + } + + public void setHasPendingWork(boolean theHasPendingWork) { + myHasPendingWork = theHasPendingWork; + } } diff --git a/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/model/ReindexJobStatus.java b/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/model/ReindexJobStatus.java new file mode 100644 index 00000000000..227756c5e44 --- /dev/null +++ b/hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/model/ReindexJobStatus.java @@ -0,0 +1,16 @@ +package ca.uhn.fhir.jpa.api.model; + +public class ReindexJobStatus { + + public static ReindexJobStatus NO_WORK_NEEDED = new ReindexJobStatus(); + + private boolean myHasReindexWorkPending; + + public boolean isHasReindexWorkPending() { + return myHasReindexWorkPending; + } + + public void setHasReindexWorkPending(boolean theHasReindexWorkPending) { + myHasReindexWorkPending = theHasReindexWorkPending; + } +}