mirror of
https://github.com/hapifhir/hapi-fhir.git
synced 2025-02-16 09:55:09 +00:00
Concurrent Transaction Conditonal Creates (#4639)
* push broken test for testing placeholder create in transaction retry * reproduce error for transaction retry * cleanup test * fix test testPlaceholderCreateTransactionRetry * revert temp fix * revert spacing * Add tests * Remove fixme * Test fix * test fix * Test fix * Remove test line * Remove unneeded change * Review comments * Add docs * Test logging * Add test logging * Add logging for intermittents --------- Co-authored-by: aditya_dave <aditya@smilecdr.com> Co-authored-by: James Agnew <jamesagnew@gmail.com>
This commit is contained in:
Aditya Dave
GitHub
parent
ea8b68f5e5
commit
8956b273f0
@ -0,0 +1,6 @@
|
||||
---
|
||||
type: fix
|
||||
issue: 4639
|
||||
title: "When executing FHIR transactions in the JPA server with automatic retry enabled, if
|
||||
automatic placeholder-reference creation is enabled the system could sometimes fail to
|
||||
automatically retry. This has been corrected."
|
||||
@ -53,9 +53,12 @@ public class JpaBatch2Config extends BaseBatch2Config {
|
||||
IJobPersistence retVal = batch2JobInstancePersister(theJobInstanceRepository, theWorkChunkRepository, theTransactionService, theEntityManager);
|
||||
// Avoid H2 synchronization issues caused by
|
||||
// https://github.com/h2database/h2database/issues/1808
|
||||
if (HapiSystemProperties.isUnitTestModeEnabled()) {
|
||||
retVal = ProxyUtil.synchronizedProxy(IJobPersistence.class, retVal);
|
||||
}
|
||||
// TODO: Update 2023-03-14 - The bug above appears to be fixed. I'm going to try
|
||||
// disabing this and see if we can get away without it. If so, we can delete
|
||||
// this entirely
|
||||
// if (HapiSystemProperties.isUnitTestModeEnabled()) {
|
||||
// retVal = ProxyUtil.synchronizedProxy(IJobPersistence.class, retVal);
|
||||
// }
|
||||
return retVal;
|
||||
}
|
||||
|
||||
|
||||
@ -326,15 +326,15 @@ public abstract class BaseHapiFhirResourceDao<T extends IBaseResource> extends B
|
||||
return myTxTemplate.execute(tx -> {
|
||||
IIdType retVal = myIdHelperService.translatePidIdToForcedId(myFhirContext, myResourceName, pid);
|
||||
if (!retVal.hasVersionIdPart()) {
|
||||
IIdType idWithVersion = myMemoryCacheService.getIfPresent(MemoryCacheService.CacheEnum.RESOURCE_CONDITIONAL_CREATE_VERSION, pid.getId());
|
||||
if (idWithVersion == null) {
|
||||
Long version = myResourceTableDao.findCurrentVersionByPid(pid.getId());
|
||||
Long version = myMemoryCacheService.getIfPresent(MemoryCacheService.CacheEnum.RESOURCE_CONDITIONAL_CREATE_VERSION, pid.getId());
|
||||
if (version == null) {
|
||||
version = myResourceTableDao.findCurrentVersionByPid(pid.getId());
|
||||
if (version != null) {
|
||||
retVal = myFhirContext.getVersion().newIdType().setParts(retVal.getBaseUrl(), retVal.getResourceType(), retVal.getIdPart(), Long.toString(version));
|
||||
myMemoryCacheService.putAfterCommit(MemoryCacheService.CacheEnum.RESOURCE_CONDITIONAL_CREATE_VERSION, pid.getId(), retVal);
|
||||
myMemoryCacheService.putAfterCommit(MemoryCacheService.CacheEnum.RESOURCE_CONDITIONAL_CREATE_VERSION, pid.getId(), version);
|
||||
}
|
||||
} else {
|
||||
retVal = idWithVersion;
|
||||
}
|
||||
if (version != null) {
|
||||
retVal = myFhirContext.getVersion().newIdType().setParts(retVal.getBaseUrl(), retVal.getResourceType(), retVal.getIdPart(), Long.toString(version));
|
||||
}
|
||||
}
|
||||
return retVal;
|
||||
|
||||
@ -28,7 +28,6 @@ public class ExtractInlineReferenceParams {
|
||||
private IBaseResource myResource;
|
||||
private TransactionDetails myTransactionDetails;
|
||||
private RequestDetails myRequestDetails;
|
||||
private boolean myFailOnInvalidReferences;
|
||||
|
||||
public ExtractInlineReferenceParams(
|
||||
IBaseResource theResource,
|
||||
@ -64,11 +63,4 @@ public class ExtractInlineReferenceParams {
|
||||
myRequestDetails = theRequestDetails;
|
||||
}
|
||||
|
||||
public boolean isFailOnInvalidReferences() {
|
||||
return myFailOnInvalidReferences;
|
||||
}
|
||||
|
||||
public void setFailOnInvalidReferences(boolean theFailOnInvalidReferences) {
|
||||
myFailOnInvalidReferences = theFailOnInvalidReferences;
|
||||
}
|
||||
}
|
||||
|
||||
@ -42,6 +42,8 @@ import ca.uhn.fhir.jpa.searchparam.extractor.ResourceIndexedSearchParams;
|
||||
import ca.uhn.fhir.jpa.searchparam.extractor.SearchParamExtractorService;
|
||||
import ca.uhn.fhir.jpa.util.MemoryCacheService;
|
||||
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.NotFoundPid;
|
||||
import ca.uhn.fhir.rest.api.server.storage.TransactionDetails;
|
||||
import ca.uhn.fhir.rest.server.exceptions.InvalidRequestException;
|
||||
import ca.uhn.fhir.rest.server.exceptions.PreconditionFailedException;
|
||||
@ -112,12 +114,13 @@ public class SearchParamWithInlineReferencesExtractor {
|
||||
mySearchParamRegistry = theSearchParamRegistry;
|
||||
}
|
||||
|
||||
public void populateFromResource(RequestPartitionId theRequestPartitionId, ResourceIndexedSearchParams theParams, TransactionDetails theTransactionDetails, ResourceTable theEntity, IBaseResource theResource, ResourceIndexedSearchParams theExistingParams, RequestDetails theRequest, boolean theFailOnInvalidReference) {
|
||||
ExtractInlineReferenceParams theExtractParams = new ExtractInlineReferenceParams(theResource, theTransactionDetails, theRequest);
|
||||
theExtractParams.setFailOnInvalidReferences(theFailOnInvalidReference);
|
||||
extractInlineReferences(theExtractParams);
|
||||
public void populateFromResource(RequestPartitionId theRequestPartitionId, ResourceIndexedSearchParams theParams, TransactionDetails theTransactionDetails, ResourceTable theEntity, IBaseResource theResource, ResourceIndexedSearchParams theExistingParams, RequestDetails theRequest, boolean thePerformIndexing) {
|
||||
if (thePerformIndexing) {
|
||||
ExtractInlineReferenceParams extractParams = new ExtractInlineReferenceParams(theResource, theTransactionDetails, theRequest);
|
||||
extractInlineReferences(extractParams);
|
||||
}
|
||||
|
||||
mySearchParamExtractorService.extractFromResource(theRequestPartitionId, theRequest, theParams, theExistingParams, theEntity, theResource, theTransactionDetails, theFailOnInvalidReference);
|
||||
mySearchParamExtractorService.extractFromResource(theRequestPartitionId, theRequest, theParams, theExistingParams, theEntity, theResource, theTransactionDetails, thePerformIndexing);
|
||||
|
||||
ResourceSearchParams activeSearchParams = mySearchParamRegistry.getActiveSearchParams(theEntity.getResourceType());
|
||||
if (myStorageSettings.getIndexMissingFields() == JpaStorageSettings.IndexEnabledEnum.ENABLED) {
|
||||
@ -242,8 +245,18 @@ public class SearchParamWithInlineReferencesExtractor {
|
||||
}
|
||||
Class<? extends IBaseResource> matchResourceType = matchResourceDef.getImplementingClass();
|
||||
|
||||
JpaPid resolvedMatch = null;
|
||||
if (theTransactionDetails != null) {
|
||||
resolvedMatch = (JpaPid) theTransactionDetails.getResolvedMatchUrls().get(nextIdText);
|
||||
}
|
||||
|
||||
//Attempt to find the target reference before creating a placeholder
|
||||
Set<JpaPid> matches = myMatchResourceUrlService.processMatchUrl(nextIdText, matchResourceType, theTransactionDetails, theRequest);
|
||||
Set<JpaPid> matches;
|
||||
if (resolvedMatch != null && !IResourcePersistentId.NOT_FOUND.equals(resolvedMatch)) {
|
||||
matches = Set.of(resolvedMatch);
|
||||
} else {
|
||||
matches = myMatchResourceUrlService.processMatchUrl(nextIdText, matchResourceType, theTransactionDetails, theRequest);
|
||||
}
|
||||
|
||||
JpaPid match;
|
||||
if (matches.isEmpty()) {
|
||||
@ -268,6 +281,10 @@ public class SearchParamWithInlineReferencesExtractor {
|
||||
IIdType newId = myIdHelperService.translatePidIdToForcedId(myContext, resourceTypeString, match);
|
||||
ourLog.debug("Replacing inline match URL[{}] with ID[{}}", nextId.getValue(), newId);
|
||||
|
||||
if (theTransactionDetails != null) {
|
||||
String previousReference = nextRef.getReferenceElement().getValue();
|
||||
theTransactionDetails.addRollbackUndoAction(()->nextRef.setReference(previousReference));
|
||||
}
|
||||
nextRef.setReference(newId.getValue());
|
||||
}
|
||||
}
|
||||
@ -308,11 +325,7 @@ public class SearchParamWithInlineReferencesExtractor {
|
||||
myEntityManager.persist(next);
|
||||
haveNewStringUniqueParams = true;
|
||||
}
|
||||
if (theParams.myComboStringUniques.size() > 0 || haveNewStringUniqueParams) {
|
||||
theEntity.setParamsComboStringUniquePresent(true);
|
||||
} else {
|
||||
theEntity.setParamsComboStringUniquePresent(false);
|
||||
}
|
||||
theEntity.setParamsComboStringUniquePresent(theParams.myComboStringUniques.size() > 0 || haveNewStringUniqueParams);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -10,9 +10,9 @@ import ca.uhn.fhir.jpa.api.dao.IFhirSystemDao;
|
||||
import ca.uhn.fhir.jpa.api.svc.IIdHelperService;
|
||||
import ca.uhn.fhir.jpa.cache.IResourceVersionSvc;
|
||||
import ca.uhn.fhir.jpa.dao.r4.TransactionProcessorVersionAdapterR4;
|
||||
import ca.uhn.fhir.jpa.dao.tx.HapiTransactionService;
|
||||
import ca.uhn.fhir.jpa.dao.tx.IHapiTransactionService;
|
||||
import ca.uhn.fhir.jpa.dao.tx.NonTransactionalHapiTransactionService;
|
||||
import ca.uhn.fhir.jpa.model.config.PartitionSettings;
|
||||
import ca.uhn.fhir.jpa.model.entity.StorageSettings;
|
||||
import ca.uhn.fhir.jpa.partition.IRequestPartitionHelperSvc;
|
||||
import ca.uhn.fhir.jpa.searchparam.MatchUrlService;
|
||||
import ca.uhn.fhir.jpa.searchparam.matcher.InMemoryResourceMatcher;
|
||||
@ -27,6 +27,7 @@ import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.Answers;
|
||||
import org.mockito.Spy;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
@ -66,8 +67,6 @@ public class TransactionProcessorTest {
|
||||
@MockBean
|
||||
private MatchResourceUrlService myMatchResourceUrlService;
|
||||
@MockBean
|
||||
private HapiTransactionService myHapiTransactionService;
|
||||
@MockBean
|
||||
private InMemoryResourceMatcher myInMemoryResourceMatcher;
|
||||
@MockBean
|
||||
private IIdHelperService myIdHelperService;
|
||||
@ -88,13 +87,7 @@ public class TransactionProcessorTest {
|
||||
|
||||
@BeforeEach
|
||||
public void before() {
|
||||
when(myHapiTransactionService.execute(any(), any(), any())).thenAnswer(t -> {
|
||||
TransactionCallback<?> callback = t.getArgument(2, TransactionCallback.class);
|
||||
return callback.doInTransaction(mock(TransactionStatus.class));
|
||||
});
|
||||
|
||||
myTransactionProcessor.setEntityManagerForUnitTest(myEntityManager);
|
||||
|
||||
when(myEntityManager.unwrap(eq(Session.class))).thenReturn(mySession);
|
||||
}
|
||||
|
||||
@ -156,5 +149,10 @@ public class TransactionProcessorTest {
|
||||
return new TransactionProcessorVersionAdapterR4();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IHapiTransactionService hapiTransactionService() {
|
||||
return new NonTransactionalHapiTransactionService();
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@ -67,8 +67,10 @@ public class FhirResourceDaoR4ConcurrentWriteTest extends BaseJpaR4Test {
|
||||
private TransactionConcurrencySemaphoreInterceptor myConcurrencySemaphoreInterceptor;
|
||||
|
||||
|
||||
@Override
|
||||
@BeforeEach
|
||||
public void before() {
|
||||
public void before() throws Exception {
|
||||
super.before();
|
||||
myExecutor = Executors.newFixedThreadPool(10);
|
||||
myRetryInterceptor = new UserRequestRetryVersionConflictsInterceptor();
|
||||
myConcurrencySemaphoreInterceptor = new TransactionConcurrencySemaphoreInterceptor(myMemoryCacheService);
|
||||
@ -132,6 +134,33 @@ public class FhirResourceDaoR4ConcurrentWriteTest extends BaseJpaR4Test {
|
||||
@Test
|
||||
public void testTransactionCreates_WithRetry() throws ExecutionException, InterruptedException {
|
||||
myInterceptorRegistry.registerInterceptor(myRetryInterceptor);
|
||||
myStorageSettings.setUniqueIndexesEnabled(true);
|
||||
|
||||
// Create a unique search parameter to enfore uniqueness
|
||||
// TODO: remove this once we have a better way to enfore these
|
||||
SearchParameter sp = new SearchParameter();
|
||||
sp.setId("SearchParameter/Practitioner-identifier");
|
||||
sp.setType(Enumerations.SearchParamType.TOKEN);
|
||||
sp.setCode("identifier");
|
||||
sp.setExpression("Practitioner.identifier");
|
||||
sp.setStatus(Enumerations.PublicationStatus.ACTIVE);
|
||||
sp.addBase("Practitioner");
|
||||
mySearchParameterDao.update(sp);
|
||||
|
||||
sp = new SearchParameter();
|
||||
sp.setId("SearchParameter/Practitioner-identifier-unique");
|
||||
sp.setType(Enumerations.SearchParamType.COMPOSITE);
|
||||
sp.setStatus(Enumerations.PublicationStatus.ACTIVE);
|
||||
sp.addBase("Practitioner");
|
||||
sp.addComponent()
|
||||
.setExpression("Practitioner")
|
||||
.setDefinition("SearchParameter/Practitioner-identifier");
|
||||
sp.addExtension()
|
||||
.setUrl(HapiExtensions.EXT_SP_UNIQUE)
|
||||
.setValue(new BooleanType(true));
|
||||
mySearchParameterDao.update(sp);
|
||||
|
||||
mySearchParamRegistry.forceRefresh();
|
||||
|
||||
AtomicInteger setCounter = new AtomicInteger(0);
|
||||
AtomicInteger fuzzCounter = new AtomicInteger(0);
|
||||
@ -162,9 +191,9 @@ public class FhirResourceDaoR4ConcurrentWriteTest extends BaseJpaR4Test {
|
||||
|
||||
assertEquals(1, counts.get("Patient"), counts.toString());
|
||||
assertEquals(1, counts.get("Observation"), counts.toString());
|
||||
assertEquals(6, myResourceLinkDao.count());
|
||||
assertEquals(6, myResourceTableDao.count());
|
||||
assertEquals(14, myResourceHistoryTableDao.count());
|
||||
assertEquals(7, myResourceLinkDao.count()); // 1 for SP, 6 for transaction
|
||||
assertEquals(8, myResourceTableDao.count()); // 2 SPs, 6 resources
|
||||
assertEquals(16, myResourceHistoryTableDao.count());
|
||||
});
|
||||
|
||||
}
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
package ca.uhn.fhir.jpa.dao.r4;
|
||||
|
||||
import ca.uhn.fhir.i18n.Msg;
|
||||
import ca.uhn.fhir.interceptor.api.Pointcut;
|
||||
import ca.uhn.fhir.jpa.api.config.JpaStorageSettings;
|
||||
import ca.uhn.fhir.jpa.dao.BaseHapiFhirDao;
|
||||
import ca.uhn.fhir.jpa.model.entity.NormalizedQuantitySearchLevel;
|
||||
@ -78,6 +79,9 @@ import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Disabled;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.CsvSource;
|
||||
import org.springframework.dao.DataIntegrityViolationException;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.TransactionStatus;
|
||||
import org.springframework.transaction.support.TransactionCallback;
|
||||
@ -97,6 +101,7 @@ import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static org.hamcrest.MatcherAssert.assertThat;
|
||||
@ -127,13 +132,16 @@ public class FhirSystemDaoR4Test extends BaseJpaR4SystemTest {
|
||||
|
||||
@AfterEach
|
||||
public void after() {
|
||||
myStorageSettings.setAllowInlineMatchUrlReferences(false);
|
||||
myStorageSettings.setAllowMultipleDelete(new JpaStorageSettings().isAllowMultipleDelete());
|
||||
myStorageSettings.setNormalizedQuantitySearchLevel(NormalizedQuantitySearchLevel.NORMALIZED_QUANTITY_SEARCH_NOT_SUPPORTED);
|
||||
myStorageSettings.setBundleBatchPoolSize(new JpaStorageSettings().getBundleBatchPoolSize());
|
||||
myStorageSettings.setBundleBatchMaxPoolSize(new JpaStorageSettings().getBundleBatchMaxPoolSize());
|
||||
myStorageSettings.setAutoCreatePlaceholderReferenceTargets(new JpaStorageSettings().isAutoCreatePlaceholderReferenceTargets());
|
||||
myStorageSettings.setAutoVersionReferenceAtPaths(new JpaStorageSettings().getAutoVersionReferenceAtPaths());
|
||||
JpaStorageSettings defaults = new JpaStorageSettings();
|
||||
myStorageSettings.setAllowInlineMatchUrlReferences(defaults.isAllowInlineMatchUrlReferences());
|
||||
myStorageSettings.setAllowMultipleDelete(defaults.isAllowMultipleDelete());
|
||||
myStorageSettings.setNormalizedQuantitySearchLevel(defaults.getNormalizedQuantitySearchLevel());
|
||||
myStorageSettings.setBundleBatchPoolSize(defaults.getBundleBatchPoolSize());
|
||||
myStorageSettings.setBundleBatchMaxPoolSize(defaults.getBundleBatchMaxPoolSize());
|
||||
myStorageSettings.setAutoCreatePlaceholderReferenceTargets(defaults.isAutoCreatePlaceholderReferenceTargets());
|
||||
myStorageSettings.setPopulateIdentifierInAutoCreatedPlaceholderReferenceTargets(defaults.isPopulateIdentifierInAutoCreatedPlaceholderReferenceTargets());
|
||||
myStorageSettings.setAutoVersionReferenceAtPaths(defaults.getAutoVersionReferenceAtPaths());
|
||||
|
||||
myFhirContext.getParserOptions().setAutoContainReferenceTargetsWithNoId(true);
|
||||
}
|
||||
|
||||
@ -1298,7 +1306,6 @@ public class FhirSystemDaoR4Test extends BaseJpaR4SystemTest {
|
||||
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testTransactionUpdateTwoResourcesWithSameId() {
|
||||
Bundle request = new Bundle();
|
||||
@ -4076,6 +4083,96 @@ public class FhirSystemDaoR4Test extends BaseJpaR4SystemTest {
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* See #4639
|
||||
*/
|
||||
@ParameterizedTest
|
||||
@CsvSource({
|
||||
// Observation.subject ref is an inline match URL that doesn't exist, so it'll be created
|
||||
"Patient?identifier=http://nothing-matching|123" + "," + "Patient?identifier=http%3A%2F%2Facme.org|ID1" + "," + "Observation|Patient|Patient",
|
||||
// Observation.subject ref is the same ref as a conditional update URL in the Bundle
|
||||
"Patient?identifier=http%3A%2F%2Facme.org|ID1" + "," + "Patient?identifier=http%3A%2F%2Facme.org|ID1" + "," + "Observation|Patient",
|
||||
// Observation.subject ref is a placeholder UUID pointing to the Patient that is being conditionally created
|
||||
"urn:uuid:8dba64a8-2aca-48fe-8b4e-8c7bf2ab695a" + "," + "Patient?identifier=http%3A%2F%2Facme.org|ID1" + "," + "Observation|Patient"
|
||||
})
|
||||
public void testPlaceholderCreateTransactionRetry_NonMatchingPlaceholderReference(String theObservationSubjectReference, String thePatientRequestUrl, String theExpectedCreatedResourceTypes) {
|
||||
// setup
|
||||
myStorageSettings.setAllowInlineMatchUrlReferences(true);
|
||||
myStorageSettings.setPopulateIdentifierInAutoCreatedPlaceholderReferenceTargets(true);
|
||||
myStorageSettings.setAutoCreatePlaceholderReferenceTargets(true);
|
||||
|
||||
mySrd.setRetry(true);
|
||||
mySrd.setMaxRetries(3);
|
||||
|
||||
Bundle bundle = new Bundle();
|
||||
bundle.setType(BundleType.TRANSACTION);
|
||||
Patient pat = new Patient();
|
||||
if (theObservationSubjectReference.startsWith("urn:")) {
|
||||
pat.setId(theObservationSubjectReference);
|
||||
}
|
||||
pat.addIdentifier().setSystem("http://acme.org").setValue("ID1");
|
||||
Observation obs = new Observation();
|
||||
obs.setSubject(new Reference(theObservationSubjectReference));
|
||||
obs.setIdentifier(List.of(new Identifier().setSystem("http://obs.org").setValue("ID2")));
|
||||
bundle.addEntry().setResource(obs)
|
||||
.getRequest()
|
||||
.setMethod(HTTPVerb.PUT)
|
||||
.setUrl("Observation?identifier=http%3A%2F%2Fobs.org|ID2");
|
||||
bundle.addEntry().setResource(pat)
|
||||
.getRequest()
|
||||
.setMethod(HTTPVerb.PUT)
|
||||
.setUrl(thePatientRequestUrl);
|
||||
|
||||
AtomicInteger countdown = new AtomicInteger(1);
|
||||
mySrdInterceptorService.registerAnonymousInterceptor(Pointcut.STORAGE_TRANSACTION_PROCESSED, ((thePointcut, theArgs) -> {
|
||||
if (countdown.get() > 0) {
|
||||
countdown.decrementAndGet();
|
||||
// fake out a tag creation error
|
||||
throw new DataIntegrityViolationException("hfj_res_tag");
|
||||
}
|
||||
}));
|
||||
|
||||
runInTransaction(()->{
|
||||
List<ResourceTable> all = myResourceTableDao.findAll();
|
||||
List<String> storedTypes = all.stream().map(ResourceTable::getResourceType).sorted().toList();
|
||||
assertThat(storedTypes, empty());
|
||||
});
|
||||
|
||||
|
||||
// execute
|
||||
Bundle output = mySystemDao.transaction(mySrd, bundle);
|
||||
|
||||
// validate
|
||||
|
||||
String observationId = new IdType(output.getEntry().get(0).getResponse().getLocation()).toUnqualifiedVersionless().getValue();
|
||||
String patientId = new IdType(output.getEntry().get(1).getResponse().getLocation()).toUnqualifiedVersionless().getValue();
|
||||
Observation observation = myObservationDao.read(new IdType(observationId), mySrd);
|
||||
String subjectReference = observation.getSubject().getReference();
|
||||
ourLog.info("Obs: {}", observationId);
|
||||
ourLog.info("Patient: {}", patientId);
|
||||
ourLog.info("Ref: {}", subjectReference);
|
||||
if (thePatientRequestUrl.equals(theObservationSubjectReference) || theObservationSubjectReference.startsWith("urn:")) {
|
||||
assertEquals(patientId, subjectReference);
|
||||
} else {
|
||||
assertNotEquals(patientId, subjectReference);
|
||||
}
|
||||
|
||||
assertEquals(0, countdown.get());
|
||||
assertEquals("201 Created", output.getEntry().get(0).getResponse().getStatus());
|
||||
assertEquals("201 Created", output.getEntry().get(1).getResponse().getStatus());
|
||||
|
||||
ourLog.info("Assigned resource IDs:\n * " + output.getEntry().stream().map(t->t.getResponse().getLocation()).collect(Collectors.joining("\n * ")));
|
||||
|
||||
myCaptureQueriesListener.logInsertQueries(t -> t.getSql(false, false).contains(" into HFJ_RESOURCE "));
|
||||
|
||||
runInTransaction(()->{
|
||||
List<ResourceTable> all = myResourceTableDao.findAll();
|
||||
List<String> storedTypes = all.stream().map(ResourceTable::getResourceType).sorted().toList();
|
||||
assertThat("Resources:\n * " + all.stream().map(t->t.toString()).collect(Collectors.joining("\n * ")), storedTypes, contains(theExpectedCreatedResourceTypes.split("\\|")));
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Per a message on the mailing list
|
||||
*/
|
||||
|
||||
@ -160,8 +160,10 @@ public class MultitenantBatchOperationR4Test extends BaseMultitenantResourceProv
|
||||
|
||||
myBatch2JobHelper.awaitJobCompletion(jobId.getValue());
|
||||
|
||||
ourLog.info("Search params: {}", mySearchParamRegistry.getActiveSearchParams("Observation").getSearchParamNames());
|
||||
logAllTokenIndexes();
|
||||
|
||||
|
||||
// validate
|
||||
runInTransaction(()->{
|
||||
long indexedSps = myResourceIndexedSearchParamTokenDao
|
||||
@ -195,13 +197,16 @@ public class MultitenantBatchOperationR4Test extends BaseMultitenantResourceProv
|
||||
|
||||
myBatch2JobHelper.awaitJobCompletion(jobId.getValue());
|
||||
|
||||
ourLog.info("Search params: {}", mySearchParamRegistry.getActiveSearchParams("Observation").getSearchParamNames());
|
||||
logAllTokenIndexes();
|
||||
|
||||
runInTransaction(()->{
|
||||
long indexedSps = myResourceIndexedSearchParamTokenDao
|
||||
.findAll()
|
||||
.stream()
|
||||
.filter(t->t.getParamName().equals("alleleName"))
|
||||
.count();
|
||||
assertEquals(3, indexedSps, ()->"Token indexes:\n * " + myResourceIndexedSearchParamTokenDao.findAll().stream().filter(t->t.getParamName().equals("alleleName")).map(ResourceIndexedSearchParamToken::toString).collect(Collectors.joining("\n * ")));
|
||||
assertEquals(3, indexedSps, ()->"Resources:\n * " + myResourceTableDao.findAll().stream().map(t->t.toString()).collect(Collectors.joining("\n * ")));
|
||||
});
|
||||
|
||||
myTenantClientInterceptor.setTenantId(DEFAULT_PARTITION_NAME);
|
||||
@ -247,6 +252,9 @@ public class MultitenantBatchOperationR4Test extends BaseMultitenantResourceProv
|
||||
myBatch2JobHelper.awaitJobCompletion(jobId.getValue());
|
||||
|
||||
// validate
|
||||
ourLog.info("Search params: {}", mySearchParamRegistry.getActiveSearchParams("Observation").getSearchParamNames());
|
||||
logAllTokenIndexes();
|
||||
|
||||
List<String> alleleObservationIds = reindexTestHelper.getAlleleObservationIds(myClient);
|
||||
// Only the one in the first tenant should be indexed
|
||||
myTenantClientInterceptor.setTenantId(TENANT_A);
|
||||
|
||||
@ -23,12 +23,23 @@ package ca.uhn.fhir.batch2.jobs.parameters;
|
||||
import ca.uhn.fhir.interceptor.model.RequestPartitionId;
|
||||
import ca.uhn.fhir.model.api.IModelJson;
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
import org.apache.commons.lang3.builder.ToStringBuilder;
|
||||
import org.apache.commons.lang3.builder.ToStringStyle;
|
||||
|
||||
import javax.validation.constraints.Pattern;
|
||||
|
||||
public class PartitionedUrl implements IModelJson {
|
||||
@Override
|
||||
public String toString() {
|
||||
ToStringBuilder b = new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE);
|
||||
b.append("partition", myRequestPartitionId);
|
||||
b.append("myUrl", myUrl);
|
||||
return b.toString();
|
||||
}
|
||||
|
||||
@JsonProperty("url")
|
||||
@Pattern(regexp = "^[A-Z][A-Za-z0-9]+\\?.*", message = "If populated, URL must be a search URL in the form '{resourceType}?[params]'") String myUrl;
|
||||
@Pattern(regexp = "^[A-Z][A-Za-z0-9]+\\?.*", message = "If populated, URL must be a search URL in the form '{resourceType}?[params]'")
|
||||
String myUrl;
|
||||
@JsonProperty("requestPartitionId")
|
||||
RequestPartitionId myRequestPartitionId;
|
||||
|
||||
|
||||
@ -85,6 +85,11 @@ public class ResourceIdListStep<PT extends PartitionedJobParameters, IT extends
|
||||
}
|
||||
|
||||
ourLog.info("Found {} IDs from {} to {}", nextChunk.size(), nextStart, nextChunk.getLastDate());
|
||||
if (nextChunk.size() < 10) {
|
||||
// TODO: I've added this in order to troubleshoot MultitenantBatchOperationR4Test
|
||||
// which is failing intermittently. If that stops, makes sense to remove this
|
||||
ourLog.info(" * PIDS: {}", nextChunk);
|
||||
}
|
||||
|
||||
for (TypedResourcePid typedResourcePid : nextChunk.getTypedResourcePids()) {
|
||||
TypedPidJson nextId = new TypedPidJson(typedResourcePid);
|
||||
|
||||
@ -202,6 +202,8 @@ public class JobMaintenanceServiceImpl implements IJobMaintenanceService, IHasSc
|
||||
}
|
||||
try {
|
||||
doMaintenancePass();
|
||||
} catch (Exception e) {
|
||||
ourLog.error("Maintenance pass failed", e);
|
||||
} finally {
|
||||
myRunMaintenanceSemaphore.release();
|
||||
}
|
||||
|
||||
@ -75,5 +75,10 @@ abstract public class BaseResourcePidList implements IResourcePidList {
|
||||
public IResourcePersistentId getId(int theIndex) {
|
||||
return myIds.get(theIndex);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return myIds.toString();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -22,6 +22,7 @@ package ca.uhn.fhir.jpa.api.pid;
|
||||
|
||||
import ca.uhn.fhir.i18n.Msg;
|
||||
import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId;
|
||||
import org.apache.commons.lang3.builder.ToStringBuilder;
|
||||
|
||||
import javax.annotation.Nonnull;
|
||||
import java.util.Collections;
|
||||
@ -62,4 +63,9 @@ public class EmptyResourcePidList implements IResourcePidList {
|
||||
public boolean isEmpty() {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "[empty]";
|
||||
}
|
||||
}
|
||||
|
||||
@ -39,6 +39,7 @@ import ca.uhn.fhir.jpa.api.model.LazyDaoMethodOutcome;
|
||||
import ca.uhn.fhir.jpa.cache.IResourceVersionSvc;
|
||||
import ca.uhn.fhir.jpa.cache.ResourcePersistentIdMap;
|
||||
import ca.uhn.fhir.jpa.dao.tx.HapiTransactionService;
|
||||
import ca.uhn.fhir.jpa.dao.tx.IHapiTransactionService;
|
||||
import ca.uhn.fhir.jpa.delete.DeleteConflictUtil;
|
||||
import ca.uhn.fhir.jpa.model.cross.IBasePersistedResource;
|
||||
import ca.uhn.fhir.jpa.model.entity.ResourceTable;
|
||||
@ -58,6 +59,7 @@ import ca.uhn.fhir.rest.api.PreferReturnEnum;
|
||||
import ca.uhn.fhir.rest.api.RestOperationTypeEnum;
|
||||
import ca.uhn.fhir.rest.api.server.RequestDetails;
|
||||
import ca.uhn.fhir.rest.api.server.storage.DeferredInterceptorBroadcasts;
|
||||
import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId;
|
||||
import ca.uhn.fhir.rest.api.server.storage.TransactionDetails;
|
||||
import ca.uhn.fhir.rest.param.ParameterUtil;
|
||||
import ca.uhn.fhir.rest.server.RestfulServerUtils;
|
||||
@ -107,20 +109,7 @@ import org.springframework.transaction.support.TransactionCallback;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
|
||||
import javax.annotation.Nonnull;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Comparator;
|
||||
import java.util.Date;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.IdentityHashMap;
|
||||
import java.util.Iterator;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
@ -151,7 +140,7 @@ public abstract class BaseTransactionProcessor {
|
||||
@Autowired
|
||||
private IInterceptorBroadcaster myInterceptorBroadcaster;
|
||||
@Autowired
|
||||
private HapiTransactionService myHapiTransactionService;
|
||||
private IHapiTransactionService myHapiTransactionService;
|
||||
@Autowired
|
||||
private StorageSettings myStorageSettings;
|
||||
@Autowired
|
||||
@ -618,7 +607,10 @@ public abstract class BaseTransactionProcessor {
|
||||
EntriesToProcessMap entriesToProcess;
|
||||
|
||||
try {
|
||||
entriesToProcess = myHapiTransactionService.execute(theRequestDetails, theTransactionDetails, txCallback);
|
||||
entriesToProcess = myHapiTransactionService
|
||||
.withRequest(theRequestDetails)
|
||||
.withTransactionDetails(theTransactionDetails)
|
||||
.execute(txCallback);
|
||||
} finally {
|
||||
if (haveWriteOperationsHooks(theRequestDetails)) {
|
||||
callWriteOperationsHook(Pointcut.STORAGE_TRANSACTION_WRITE_OPERATIONS_POST, theRequestDetails, theTransactionDetails, writeOperationsDetails);
|
||||
@ -920,6 +912,11 @@ public abstract class BaseTransactionProcessor {
|
||||
|
||||
theTransactionStopWatch.startTask("Bundle.entry[" + i + "]: " + verb + " " + defaultString(resourceType));
|
||||
|
||||
if (res != null) {
|
||||
String previousResourceId = res.getIdElement().getValue();
|
||||
theTransactionDetails.addRollbackUndoAction(() -> res.setId(previousResourceId));
|
||||
}
|
||||
|
||||
switch (verb) {
|
||||
case "POST": {
|
||||
// CREATE
|
||||
|
||||
@ -177,6 +177,10 @@ public class DaoResourceLinkResolver<T extends IResourcePersistentId> implements
|
||||
}
|
||||
|
||||
if (theIdToAssignToPlaceholder != null) {
|
||||
if (theTransactionDetails != null) {
|
||||
String existingId = newResource.getIdElement().getValue();
|
||||
theTransactionDetails.addRollbackUndoAction(() -> newResource.setId(existingId));
|
||||
}
|
||||
newResource.setId(resName + "/" + theIdToAssignToPlaceholder);
|
||||
valueOf = placeholderResourceDao.update(newResource, theRequest).getEntity();
|
||||
} else {
|
||||
|
||||
@ -37,6 +37,7 @@ import java.util.Date;
|
||||
import java.util.List;
|
||||
import java.util.Queue;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.function.Predicate;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
@ -320,9 +321,17 @@ public class CircularQueueCaptureQueriesListener extends BaseCaptureQueriesListe
|
||||
* Log all captured INSERT queries
|
||||
*/
|
||||
public int logInsertQueries() {
|
||||
return logInsertQueries(null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Log all captured INSERT queries
|
||||
*/
|
||||
public int logInsertQueries(Predicate<SqlQuery> theInclusionPredicate) {
|
||||
List<SqlQuery> insertQueries = getInsertQueries();
|
||||
List<String> queries = insertQueries
|
||||
.stream()
|
||||
.filter(t -> theInclusionPredicate == null || theInclusionPredicate.test(t))
|
||||
.map(CircularQueueCaptureQueriesListener::formatQueryAsSql)
|
||||
.collect(Collectors.toList());
|
||||
ourLog.info("Insert Queries:\n{}", String.join("\n", queries));
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user