From 2e030eebaa86a44f1f893ed06bbad9170d0b2430 Mon Sep 17 00:00:00 2001 From: James Agnew Date: Wed, 21 Nov 2018 12:06:10 -0500 Subject: [PATCH] Make sure that reindexing happens correctly --- .../index/SearchParamExtractorService.java | 5 +- .../jpa/entity/ResourceReindexJobEntity.java | 22 +++- .../jpa/search/SearchCoordinatorSvcImpl.java | 7 +- .../reindex/IResourceReindexingSvc.java | 10 +- .../reindex/ResourceReindexingSvcImpl.java | 58 +++++++--- .../subscription/ResourceDeliveryMessage.java | 5 +- ...scriptionDeliveringRestHookSubscriber.java | 11 +- .../dao/r4/FhirResourceDaoR4DeleteTest.java | 109 ++++++++++++++++++ ...hirResourceDaoR4UniqueSearchParamTest.java | 13 +++ .../ResourceReindexingSvcImplTest.java | 53 +++++++-- .../uhn/fhir/rest/server/RestfulServer.java | 93 +++++++-------- src/changes/changes.xml | 5 + 12 files changed, 302 insertions(+), 89 deletions(-) create mode 100644 hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4DeleteTest.java diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/index/SearchParamExtractorService.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/index/SearchParamExtractorService.java index cd8f486df02..35057876b16 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/index/SearchParamExtractorService.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/index/SearchParamExtractorService.java @@ -214,7 +214,8 @@ public class SearchParamExtractorService { private void extractCompositeStringUniques(ResourceTable theEntity, ResourceIndexedSearchParams theParams) { - List uniqueSearchParams = mySearchParamRegistry.getActiveUniqueSearchParams(theEntity.getResourceType()); + String resourceType = theEntity.getResourceType(); + List uniqueSearchParams = mySearchParamRegistry.getActiveUniqueSearchParams(resourceType); for (JpaRuntimeSearchParam next : uniqueSearchParams) { @@ -282,7 +283,7 @@ public class SearchParamExtractorService { } } - Set queryStringsToPopulate = theParams.extractCompositeStringUniquesValueChains(theEntity.getResourceType(), partsChoices); + Set queryStringsToPopulate = theParams.extractCompositeStringUniquesValueChains(resourceType, partsChoices); for (String nextQueryString : queryStringsToPopulate) { if (isNotBlank(nextQueryString)) { diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/entity/ResourceReindexJobEntity.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/entity/ResourceReindexJobEntity.java index 9676d2f1aa8..1fef31d7490 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/entity/ResourceReindexJobEntity.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/entity/ResourceReindexJobEntity.java @@ -9,9 +9,9 @@ package ca.uhn.fhir.jpa.entity; * 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. @@ -21,6 +21,8 @@ package ca.uhn.fhir.jpa.entity; */ import com.google.common.annotations.VisibleForTesting; +import org.apache.commons.lang3.builder.ToStringBuilder; +import org.apache.commons.lang3.builder.ToStringStyle; import javax.persistence.*; import java.io.Serializable; @@ -110,4 +112,20 @@ public class ResourceReindexJobEntity implements Serializable { public void setDeleted(boolean theDeleted) { myDeleted = theDeleted; } + + @Override + public String toString() { + ToStringBuilder b = new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE) + .append("id", myId) + .append("resourceType", myResourceType) + .append("thresholdLow", myThresholdLow) + .append("thresholdHigh", myThresholdHigh); + if (myDeleted) { + b.append("deleted", myDeleted); + } + if (mySuspendedUntil != null) { + b.append("suspendedUntil", mySuspendedUntil); + } + return b.toString(); + } } diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/SearchCoordinatorSvcImpl.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/SearchCoordinatorSvcImpl.java index 00ab1343f11..7316bf0608d 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/SearchCoordinatorSvcImpl.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/SearchCoordinatorSvcImpl.java @@ -245,8 +245,7 @@ public class SearchCoordinatorSvcImpl implements ISearchCoordinatorSvc { }); } catch (Exception e) { ourLog.warn("Failed to activate search: {}", e.toString()); - // FIXME: aaaaa - ourLog.info("Failed to activate search", e); + ourLog.trace("Failed to activate search", e); return Optional.empty(); } } @@ -518,10 +517,6 @@ public class SearchCoordinatorSvcImpl implements ISearchCoordinatorSvc { * user has requested resources 0-60, then they would get 0-50 back but the search * coordinator would then stop searching.SearchCoordinatorSvcImplTest */ - // FIXME: aaaaaaaa -// List remainingResources = SearchCoordinatorSvcImpl.this.getResources(mySearch.getUuid(), mySyncedPids.size(), theToIndex); -// ourLog.debug("Adding {} resources to the existing {} synced resource IDs", remainingResources.size(), mySyncedPids.size()); -// mySyncedPids.addAll(remainingResources); keepWaiting = false; break; case FAILED: diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/reindex/IResourceReindexingSvc.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/reindex/IResourceReindexingSvc.java index 571a78fee8f..e5d851ca795 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/reindex/IResourceReindexingSvc.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/reindex/IResourceReindexingSvc.java @@ -20,6 +20,11 @@ package ca.uhn.fhir.jpa.search.reindex; * #L% */ +import org.apache.commons.lang3.time.DateUtils; +import org.springframework.scheduling.annotation.Scheduled; + +import javax.transaction.Transactional; + public interface IResourceReindexingSvc { /** @@ -34,7 +39,10 @@ public interface IResourceReindexingSvc { /** * Called automatically by the job scheduler - * + */ + void scheduleReindexingPass(); + + /** * @return Returns null if the system did not attempt to perform a pass because one was * already proceeding. Otherwise, returns the number of resources affected. */ diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/reindex/ResourceReindexingSvcImpl.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/reindex/ResourceReindexingSvcImpl.java index 8f52ac36012..0eb0e366960 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/reindex/ResourceReindexingSvcImpl.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/search/reindex/ResourceReindexingSvcImpl.java @@ -9,9 +9,9 @@ package ca.uhn.fhir.jpa.search.reindex; * 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. @@ -36,6 +36,7 @@ import ca.uhn.fhir.rest.server.exceptions.InternalErrorException; import ca.uhn.fhir.rest.server.exceptions.ResourceVersionConflictException; import ca.uhn.fhir.util.StopWatch; import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Stopwatch; import org.apache.commons.lang3.Validate; import org.apache.commons.lang3.concurrent.BasicThreadFactory; import org.apache.commons.lang3.time.DateUtils; @@ -71,8 +72,9 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank; public class ResourceReindexingSvcImpl implements IResourceReindexingSvc { - private static final Date BEGINNING_OF_TIME = new Date(0); + static final Date BEGINNING_OF_TIME = new Date(0); private static final Logger ourLog = LoggerFactory.getLogger(ResourceReindexingSvcImpl.class); + public static final int PASS_SIZE = 25000; private final ReentrantLock myIndexingLock = new ReentrantLock(); @Autowired private IResourceReindexJobDao myReindexJobDao; @@ -176,6 +178,13 @@ public class ResourceReindexingSvcImpl implements IResourceReindexingSvc { @Override @Transactional(Transactional.TxType.NEVER) @Scheduled(fixedDelay = 10 * DateUtils.MILLIS_PER_SECOND) + public void scheduleReindexingPass() { + runReindexingPass(); + } + + + @Override + @Transactional(Transactional.TxType.NEVER) public Integer runReindexingPass() { if (myDaoConfig.isSchedulingDisabled()) { return null; @@ -223,10 +232,16 @@ public class ResourceReindexingSvcImpl implements IResourceReindexingSvc { Collection jobs = myTxTemplate.execute(t -> myReindexJobDao.findAll(PageRequest.of(0, 10), false)); assert jobs != null; + if (jobs.size() > 0) { + ourLog.info("Running {} reindex jobs: {}", jobs.size(), jobs); + } else { + ourLog.debug("Running {} reindex jobs: {}", jobs.size(), jobs); + } + int count = 0; for (ResourceReindexJobEntity next : jobs) { - if (next.getThresholdHigh().getTime() < System.currentTimeMillis()) { + if (next.getThresholdLow() != null && next.getThresholdLow().getTime() >= next.getThresholdHigh().getTime()) { markJobAsDeleted(next); continue; } @@ -236,9 +251,10 @@ public class ResourceReindexingSvcImpl implements IResourceReindexingSvc { return count; } - private void markJobAsDeleted(ResourceReindexJobEntity next) { + private void markJobAsDeleted(ResourceReindexJobEntity theJob) { + ourLog.info("Marking reindexing job ID[{}] as deleted", theJob.getId()); myTxTemplate.execute(t -> { - myReindexJobDao.markAsDeletedById(next.getId()); + myReindexJobDao.markAsDeletedById(theJob.getId()); return null; }); } @@ -259,8 +275,9 @@ public class ResourceReindexingSvcImpl implements IResourceReindexingSvc { Date high = theJob.getThresholdHigh(); // Query for resources within threshold + StopWatch pageSw = new StopWatch(); Slice range = myTxTemplate.execute(t -> { - PageRequest page = PageRequest.of(0, 10000); + PageRequest page = PageRequest.of(0, PASS_SIZE); if (isNotBlank(theJob.getResourceType())) { return myResourceTableDao.findIdsOfResourcesWithinUpdatedRangeOrderedFromOldest(page, theJob.getResourceType(), low, high); } else { @@ -269,6 +286,13 @@ public class ResourceReindexingSvcImpl implements IResourceReindexingSvc { }); Validate.notNull(range); int count = range.getNumberOfElements(); + ourLog.info("Loaded {} resources for reindexing in {}", count, pageSw.toString()); + + // If we didn't find any results at all, mark as deleted + if (count == 0) { + markJobAsDeleted(theJob); + return 0; + } // Submit each resource requiring reindexing List> futures = range @@ -304,18 +328,15 @@ public class ResourceReindexingSvcImpl implements IResourceReindexingSvc { } } - // Just in case we end up in some sort of infinite loop. This shouldn't happen, and couldn't really - // happen unless there were 10000 resources with the exact same update time down to the - // millisecond. Date newLow; - if (latestDate == null) { - markJobAsDeleted(theJob); - return 0; - } if (latestDate.getTime() == low.getTime()) { - ourLog.error("Final pass time for reindex JOB[{}] has same ending low value: {}", theJob.getId(), latestDate); - newLow = new Date(latestDate.getTime() + 1); - } else if (!haveMultipleDates) { + if (count == PASS_SIZE) { + // Just in case we end up in some sort of infinite loop. This shouldn't happen, and couldn't really + // happen unless there were 10000 resources with the exact same update time down to the + // millisecond. + ourLog.error("Final pass time for reindex JOB[{}] has same ending low value: {}", theJob.getId(), latestDate); + } + newLow = new Date(latestDate.getTime() + 1); } else { newLow = latestDate; @@ -326,7 +347,7 @@ public class ResourceReindexingSvcImpl implements IResourceReindexingSvc { return null; }); - ourLog.info("Completed pass of reindex JOB[{}] - Indexed {} resources in {} ({} / sec) - Have indexed until: {}", theJob.getId(), count, sw.toString(), sw.formatThroughput(count, TimeUnit.SECONDS), theJob.getThresholdLow()); + ourLog.info("Completed pass of reindex JOB[{}] - Indexed {} resources in {} ({} / sec) - Have indexed until: {}", theJob.getId(), count, sw.toString(), sw.formatThroughput(count, TimeUnit.SECONDS), newLow); return counter.get(); } @@ -450,6 +471,7 @@ public class ResourceReindexingSvcImpl implements IResourceReindexingSvc { return e; } }); + } catch (ResourceVersionConflictException e) { /* * We reindex in multiple threads, so it's technically possible that two threads try diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/ResourceDeliveryMessage.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/ResourceDeliveryMessage.java index 9191ff186a1..e4a495d4bc0 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/ResourceDeliveryMessage.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/ResourceDeliveryMessage.java @@ -27,6 +27,8 @@ import org.apache.commons.lang3.Validate; import org.hl7.fhir.instance.model.api.IBaseResource; import org.hl7.fhir.instance.model.api.IIdType; +import static org.apache.commons.lang3.StringUtils.isNotBlank; + @JsonInclude(JsonInclude.Include.NON_NULL) @JsonAutoDetect(creatorVisibility = JsonAutoDetect.Visibility.NONE, fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE) public class ResourceDeliveryMessage { @@ -62,9 +64,8 @@ public class ResourceDeliveryMessage { } public IBaseResource getPayload(FhirContext theCtx) { - Validate.notNull(myPayloadString); IBaseResource retVal = myPayload; - if (retVal == null) { + if (retVal == null && isNotBlank(myPayloadString)) { retVal = theCtx.newJsonParser().parseResource(myPayloadString); myPayload = retVal; } diff --git a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/resthook/SubscriptionDeliveringRestHookSubscriber.java b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/resthook/SubscriptionDeliveringRestHookSubscriber.java index f4febfec04a..dd34f1d5abd 100644 --- a/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/resthook/SubscriptionDeliveringRestHookSubscriber.java +++ b/hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/resthook/SubscriptionDeliveringRestHookSubscriber.java @@ -21,6 +21,7 @@ package ca.uhn.fhir.jpa.subscription.resthook; */ import ca.uhn.fhir.context.FhirContext; +import ca.uhn.fhir.context.RuntimeResourceDefinition; import ca.uhn.fhir.jpa.dao.IFhirResourceDao; import ca.uhn.fhir.jpa.subscription.BaseSubscriptionDeliverySubscriber; import ca.uhn.fhir.jpa.subscription.BaseSubscriptionInterceptor; @@ -129,12 +130,14 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDe protected IBaseResource getAndMassagePayload(ResourceDeliveryMessage theMsg, CanonicalSubscription theSubscription) { IBaseResource payloadResource = theMsg.getPayload(getContext()); - if (theSubscription.getRestHookDetails().isDeliverLatestVersion()) { - IFhirResourceDao dao = getSubscriptionInterceptor().getDao(payloadResource.getClass()); + if (payloadResource == null || theSubscription.getRestHookDetails().isDeliverLatestVersion()) { + IIdType payloadId = theMsg.getPayloadId(getContext()); + RuntimeResourceDefinition resourceDef = getContext().getResourceDefinition(payloadId.getResourceType()); + IFhirResourceDao dao = getSubscriptionInterceptor().getDao(resourceDef.getImplementingClass()); try { - payloadResource = dao.read(payloadResource.getIdElement().toVersionless()); + payloadResource = dao.read(payloadId.toVersionless()); } catch (ResourceGoneException e) { - ourLog.warn("Resource {} is deleted, not going to deliver for subscription {}", payloadResource.getIdElement(), theSubscription.getIdElement(getContext())); + ourLog.warn("Resource {} is deleted, not going to deliver for subscription {}", payloadId.toVersionless(), theSubscription.getIdElement(getContext())); return null; } } diff --git a/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4DeleteTest.java b/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4DeleteTest.java new file mode 100644 index 00000000000..8170d45dcb4 --- /dev/null +++ b/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4DeleteTest.java @@ -0,0 +1,109 @@ +package ca.uhn.fhir.jpa.dao.r4; + +import ca.uhn.fhir.jpa.entity.ResourceHistoryTable; +import ca.uhn.fhir.jpa.entity.ResourceTable; +import ca.uhn.fhir.rest.server.exceptions.ResourceGoneException; +import ca.uhn.fhir.util.TestUtil; +import org.hl7.fhir.instance.model.api.IIdType; +import org.hl7.fhir.r4.model.*; +import org.junit.AfterClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.junit.Assert.*; + +public class FhirResourceDaoR4DeleteTest extends BaseJpaR4Test { + private static final Logger ourLog = LoggerFactory.getLogger(FhirResourceDaoR4DeleteTest.class); + + @Test + public void testDeleteMarksResourceAndVersionAsDeleted() { + + Patient p = new Patient(); + p.setActive(true); + IIdType id = myPatientDao.create(p).getId().toUnqualifiedVersionless(); + + myPatientDao.delete(id); + + // Table should be marked as deleted + runInTransaction(()->{ + ResourceTable resourceTable = myResourceTableDao.findById(id.getIdPartAsLong()).get(); + assertNotNull(resourceTable.getDeleted()); + }); + + // Current version should be marked as deleted + runInTransaction(()->{ + ResourceHistoryTable resourceTable = myResourceHistoryTableDao.findForIdAndVersion(id.getIdPartAsLong(), 1); + assertNull(resourceTable.getDeleted()); + }); + runInTransaction(()->{ + ResourceHistoryTable resourceTable = myResourceHistoryTableDao.findForIdAndVersion(id.getIdPartAsLong(), 2); + assertNotNull(resourceTable.getDeleted()); + }); + + try { + myPatientDao.read(id.toUnqualifiedVersionless()); + fail(); + } catch (ResourceGoneException e) { + // good + } + + myPatientDao.read(id.toUnqualifiedVersionless().withVersion("1")); + + try { + myPatientDao.read(id.toUnqualifiedVersionless().withVersion("2")); + fail(); + } catch (ResourceGoneException e) { + // good + } + + + } + + @Test + public void testResourceIsConsideredDeletedIfOnlyResourceTableEntryIsDeleted() { + + Patient p = new Patient(); + p.setActive(true); + IIdType id = myPatientDao.create(p).getId().toUnqualifiedVersionless(); + + myPatientDao.delete(id); + + // Table should be marked as deleted + runInTransaction(()->{ + ResourceTable resourceTable = myResourceTableDao.findById(id.getIdPartAsLong()).get(); + assertNotNull(resourceTable.getDeleted()); + }); + + // Mark the current history version as not-deleted even though the actual resource + // table entry is marked deleted + runInTransaction(()->{ + ResourceHistoryTable resourceTable = myResourceHistoryTableDao.findForIdAndVersion(id.getIdPartAsLong(), 2); + resourceTable.setDeleted(null); + myResourceHistoryTableDao.save(resourceTable); + }); + + try { + myPatientDao.read(id.toUnqualifiedVersionless()); + fail(); + } catch (ResourceGoneException e) { + // good + } + + myPatientDao.read(id.toUnqualifiedVersionless().withVersion("1")); + + try { + myPatientDao.read(id.toUnqualifiedVersionless().withVersion("2")); + fail(); + } catch (ResourceGoneException e) { + // good + } + + } + + @AfterClass + public static void afterClassClearContext() { + TestUtil.clearAllStaticFieldsForUnitTest(); + } + +} diff --git a/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4UniqueSearchParamTest.java b/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4UniqueSearchParamTest.java index a1b32067a3e..6437004e4d7 100644 --- a/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4UniqueSearchParamTest.java +++ b/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/dao/r4/FhirResourceDaoR4UniqueSearchParamTest.java @@ -1,6 +1,7 @@ package ca.uhn.fhir.jpa.dao.r4; import ca.uhn.fhir.jpa.dao.DaoConfig; +import ca.uhn.fhir.jpa.dao.ISearchParamRegistry; import ca.uhn.fhir.jpa.dao.SearchBuilder; import ca.uhn.fhir.jpa.dao.SearchParameterMap; import ca.uhn.fhir.jpa.entity.ResourceIndexedCompositeStringUnique; @@ -12,6 +13,7 @@ import ca.uhn.fhir.rest.param.TokenParam; import ca.uhn.fhir.rest.server.exceptions.PreconditionFailedException; import ca.uhn.fhir.rest.server.exceptions.ResourceVersionConflictException; import ca.uhn.fhir.util.TestUtil; +import com.google.common.collect.Sets; import org.hl7.fhir.instance.model.api.IIdType; import org.hl7.fhir.r4.model.*; import org.hl7.fhir.r4.model.Enumerations.PublicationStatus; @@ -19,6 +21,7 @@ import org.junit.After; import org.junit.AfterClass; import org.junit.Before; import org.junit.Test; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.TestPropertySource; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallbackWithoutResult; @@ -47,12 +50,14 @@ public class FhirResourceDaoR4UniqueSearchParamTest extends BaseJpaR4Test { myDaoConfig.setDefaultSearchParamsCanBeOverridden(new DaoConfig().isDefaultSearchParamsCanBeOverridden()); myDaoConfig.setUniqueIndexesCheckedBeforeSave(new DaoConfig().isUniqueIndexesCheckedBeforeSave()); myDaoConfig.setSchedulingDisabled(new DaoConfig().isSchedulingDisabled()); + myDaoConfig.setUniqueIndexesEnabled(new DaoConfig().isUniqueIndexesEnabled()); } @Before public void before() { myDaoConfig.setDefaultSearchParamsCanBeOverridden(true); myDaoConfig.setSchedulingDisabled(true); + myDaoConfig.setUniqueIndexesEnabled(true); SearchBuilder.resetLastHandlerMechanismForUnitTest(); } @@ -419,6 +424,10 @@ public class FhirResourceDaoR4UniqueSearchParamTest extends BaseJpaR4Test { } + @Autowired + private ISearchParamRegistry mySearchParamRegistry; + + @Test public void testDuplicateUniqueValuesAreReIndexed() { myDaoConfig.setSchedulingDisabled(true); @@ -449,6 +458,10 @@ public class FhirResourceDaoR4UniqueSearchParamTest extends BaseJpaR4Test { createUniqueObservationSubjectDateCode(); + List uniqueSearchParams = mySearchParamRegistry.getActiveUniqueSearchParams("Observation"); + assertEquals(1, uniqueSearchParams.size()); + assertEquals(3, uniqueSearchParams.get(0).getComponents().size()); + myResourceReindexingSvc.markAllResourcesForReindexing(); myResourceReindexingSvc.forceReindexingPass(); myResourceReindexingSvc.forceReindexingPass(); diff --git a/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/search/reindex/ResourceReindexingSvcImplTest.java b/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/search/reindex/ResourceReindexingSvcImplTest.java index 1549aea1b6a..62565d1a6dc 100644 --- a/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/search/reindex/ResourceReindexingSvcImplTest.java +++ b/hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/search/reindex/ResourceReindexingSvcImplTest.java @@ -90,21 +90,52 @@ public class ResourceReindexingSvcImplTest extends BaseJpaTest { } @Test - public void testMarkJobsPastThresholdAsDeleted() { + public void testReindexPassOnlyReturnsValuesAtLowThreshold() { mockNothingToExpunge(); mockSingleReindexingJob(null); - mockFourResourcesNeedReindexing(); mockFetchFourResources(); + mockFinalResourceNeedsReindexing(); - mySingleJob.setThresholdHigh(DateUtils.addMinutes(new Date(), -1)); + mySingleJob.setThresholdLow(new Date(40 * DateUtils.MILLIS_PER_DAY)); + Date highThreshold = DateUtils.addMinutes(new Date(), -1); + mySingleJob.setThresholdHigh(highThreshold); + // Run the second pass, which should index no resources (meaning it's time to mark as deleted) mySvc.forceReindexingPass(); - - verify(myResourceTableDao, never()).findIdsOfResourcesWithinUpdatedRangeOrderedFromOldest(any(), any(), any()); verify(myResourceTableDao, never()).findIdsOfResourcesWithinUpdatedRangeOrderedFromOldest(any(), any(), any(), any()); - verify(myReindexJobDao, times(1)).markAsDeletedById(myIdCaptor.capture()); + verify(myReindexJobDao, never()).markAsDeletedById(any()); + verify(myResourceTableDao, times(1)).findIdsOfResourcesWithinUpdatedRangeOrderedFromOldest(myPageRequestCaptor.capture(), myLowCaptor.capture(), myHighCaptor.capture()); + assertEquals(new Date(40 * DateUtils.MILLIS_PER_DAY), myLowCaptor.getAllValues().get(0)); + assertEquals(highThreshold, myHighCaptor.getAllValues().get(0)); - assertEquals(123L, myIdCaptor.getValue().longValue()); + // Should mark the low threshold as 1 milli higher than the ne returned item + verify(myReindexJobDao, times(1)).setThresholdLow(eq(123L), eq(new Date((40 * DateUtils.MILLIS_PER_DAY)+1L))); + } + + @Test + public void testMarkAsDeletedIfNothingIndexed() { + mockNothingToExpunge(); + mockSingleReindexingJob(null); + mockFetchFourResources(); + // Mock resource fetch + List values = Collections.emptyList(); + when(myResourceTableDao.findIdsOfResourcesWithinUpdatedRangeOrderedFromOldest(any(),any(),any())).thenReturn(new SliceImpl<>(values)); + + mySingleJob.setThresholdLow(new Date(40 * DateUtils.MILLIS_PER_DAY)); + Date highThreshold = DateUtils.addMinutes(new Date(), -1); + mySingleJob.setThresholdHigh(highThreshold); + + // Run the second pass, which should index no resources (meaning it's time to mark as deleted) + mySvc.forceReindexingPass(); + verify(myResourceTableDao, never()).findIdsOfResourcesWithinUpdatedRangeOrderedFromOldest(any(), any(), any(), any()); + verify(myResourceTableDao, times(1)).findIdsOfResourcesWithinUpdatedRangeOrderedFromOldest(myPageRequestCaptor.capture(), myLowCaptor.capture(), myHighCaptor.capture()); + assertEquals(new Date(40 * DateUtils.MILLIS_PER_DAY), myLowCaptor.getAllValues().get(0)); + assertEquals(highThreshold, myHighCaptor.getAllValues().get(0)); + + // This time we shouldn't update the threshold + verify(myReindexJobDao, never()).setThresholdLow(any(),any()); + + verify(myReindexJobDao, times(1)).markAsDeletedById(eq(123L)); } @Test @@ -243,7 +274,13 @@ public class ResourceReindexingSvcImplTest extends BaseJpaTest { private void mockFourResourcesNeedReindexing() { // Mock resource fetch List values = Arrays.asList(0L, 1L, 2L, 3L); - when(myResourceTableDao.findIdsOfResourcesWithinUpdatedRangeOrderedFromOldest(myPageRequestCaptor.capture(), myLowCaptor.capture(), myHighCaptor.capture())).thenReturn(new SliceImpl<>(values)); + when(myResourceTableDao.findIdsOfResourcesWithinUpdatedRangeOrderedFromOldest(any(),any(),any())).thenReturn(new SliceImpl<>(values)); + } + + private void mockFinalResourceNeedsReindexing() { + // Mock resource fetch + List values = Arrays.asList(2L); // the second-last one has the highest time + when(myResourceTableDao.findIdsOfResourcesWithinUpdatedRangeOrderedFromOldest(any(),any(),any())).thenReturn(new SliceImpl<>(values)); } private void mockSingleReindexingJob(String theResourceType) { diff --git a/hapi-fhir-server/src/main/java/ca/uhn/fhir/rest/server/RestfulServer.java b/hapi-fhir-server/src/main/java/ca/uhn/fhir/rest/server/RestfulServer.java index d24e27250ff..4a73f65dea6 100644 --- a/hapi-fhir-server/src/main/java/ca/uhn/fhir/rest/server/RestfulServer.java +++ b/hapi-fhir-server/src/main/java/ca/uhn/fhir/rest/server/RestfulServer.java @@ -9,9 +9,9 @@ package ca.uhn.fhir.rest.server; * 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. @@ -346,20 +346,6 @@ public class RestfulServer extends HttpServlet implements IRestfulServer theList) { + myInterceptors.clear(); + if (theList != null) { + myInterceptors.addAll(theList); + } + } + /** * Sets (or clears) the list of interceptors * @@ -578,18 +576,6 @@ public class RestfulServer extends HttpServlet implements IRestfulServer theList) { - myInterceptors.clear(); - if (theList != null) { - myInterceptors.addAll(theList); - } - } - @Override public IPagingProvider getPagingProvider() { return myPagingProvider; @@ -616,13 +602,8 @@ public class RestfulServer extends HttpServlet implements IRestfulServer theProviders) { - Validate.noNullElements(theProviders, "theProviders must not contain any null elements"); - - myPlainProviders.clear(); - if (theProviders != null) { - myPlainProviders.addAll(theProviders); - } + public void setPlainProviders(Object... theProv) { + setPlainProviders(Arrays.asList(theProv)); } /** @@ -630,8 +611,13 @@ public class RestfulServer extends HttpServlet implements IRestfulServer theProviders) { + Validate.noNullElements(theProviders, "theProviders must not contain any null elements"); + + myPlainProviders.clear(); + if (theProviders != null) { + myPlainProviders.addAll(theProviders); + } } /** @@ -643,7 +629,8 @@ public class RestfulServer extends HttpServlet implements IRestfulServer theProviders) { - Validate.noNullElements(theProviders, "theProviders must not contain any null elements"); - + public void setResourceProviders(IResourceProvider... theResourceProviders) { myResourceProviders.clear(); - if (theProviders != null) { - myResourceProviders.addAll(theProviders); + if (theResourceProviders != null) { + myResourceProviders.addAll(Arrays.asList(theResourceProviders)); } } /** * Sets the resource providers for this server */ - public void setResourceProviders(IResourceProvider... theResourceProviders) { + public void setResourceProviders(Collection theProviders) { + Validate.noNullElements(theProviders, "theProviders must not contain any null elements"); + myResourceProviders.clear(); - if (theResourceProviders != null) { - myResourceProviders.addAll(Arrays.asList(theResourceProviders)); + if (theProviders != null) { + myResourceProviders.addAll(theProviders); } } @@ -1648,6 +1635,20 @@ public class RestfulServer extends HttpServlet implements IRestfulServer + + A bug in the JPA resource reindexer was fixed: In many cases the reindexer would + mark reindexing jobs as deleted before they had actually completed, leading to + some resources not actually being reindexed. +