Skip to content

Commit

Permalink
Refactor getOrCreateTag method to prevent transaction suspension for …
Browse files Browse the repository at this point in the history 
…XA transaction compatibility

    The getOrCreateTag method previously used a propagation behavior that caused issues with
    XA transactions when using the PostgreSQL JDBC driver. The PGXAConnection does not support
    transaction suspend/resume, which made it incompatible with the existing propagation strategy
    'PROPAGATION_REQUIRES_NEW'.

    This refactor changes the getOrCreateTag logic to perform a lookup/write in a new transaction
    as before, but running in a separate thread, such that the main transaction is not suspended.
    The result is retrieved through a future.

    This change aims to improve compatibility and prevent transaction-related issues when using HAPI-FHIR with
    XA transactions and PostgreSQL.

    Closes hapifhir#3412
  • Loading branch information
@iyt-trifork
iyt-trifork committed Aug 21, 2024
1 parent ab19930 commit 7534eb8
Show file tree
Hide file tree
Showing 2 changed files with 251 additions and 136 deletions.
146 changes: 10 additions & 136 deletions hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/BaseHapiFhirDao.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,6 @@
import ca.uhn.fhir.rest.api.RestOperationTypeEnum;
import ca.uhn.fhir.rest.api.server.RequestDetails;
import ca.uhn.fhir.rest.api.server.storage.TransactionDetails;
import ca.uhn.fhir.rest.server.exceptions.InternalErrorException;
import ca.uhn.fhir.rest.server.exceptions.InvalidRequestException;
import ca.uhn.fhir.rest.server.exceptions.UnprocessableEntityException;
import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
Expand All @@ -107,14 +106,9 @@
import jakarta.annotation.Nullable;
import jakarta.annotation.PostConstruct;
import jakarta.persistence.EntityManager;
import jakarta.persistence.NoResultException;
import jakarta.persistence.EntityManagerFactory;
import jakarta.persistence.PersistenceContext;
import jakarta.persistence.PersistenceContextType;
import jakarta.persistence.TypedQuery;
import jakarta.persistence.criteria.CriteriaBuilder;
import jakarta.persistence.criteria.CriteriaQuery;
import jakarta.persistence.criteria.Predicate;
import jakarta.persistence.criteria.Root;
import org.apache.commons.lang3.NotImplementedException;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
Expand All @@ -137,12 +131,8 @@
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.ArrayList;
import java.util.Collection;
Expand All @@ -158,7 +148,6 @@
import javax.xml.stream.events.Characters;
import javax.xml.stream.events.XMLEvent;

import static java.util.Objects.isNull;
import static java.util.Objects.nonNull;
import static org.apache.commons.collections4.CollectionUtils.isEqualCollection;
import static org.apache.commons.lang3.StringUtils.isBlank;
Expand All @@ -182,8 +171,6 @@ public abstract class BaseHapiFhirDao<T extends IBaseResource> extends BaseStora
public static final long INDEX_STATUS_INDEXED = 1L;
public static final long INDEX_STATUS_INDEXING_FAILED = 2L;
public static final String NS_JPA_PROFILE = "https://github.com/hapifhir/hapi-fhir/ns/jpa/profile";
// total attempts to do a tag transaction
private static final int TOTAL_TAG_READ_ATTEMPTS = 10;
private static final Logger ourLog = LoggerFactory.getLogger(BaseHapiFhirDao.class);
private static boolean ourValidationDisabledForUnitTest;
private static boolean ourDisableIncrementOnUpdateForUnitTest = false;
Expand Down Expand Up @@ -254,12 +241,16 @@ public abstract class BaseHapiFhirDao<T extends IBaseResource> extends BaseStora
@Autowired(required = false)
private IFulltextSearchSvc myFulltextSearchSvc;

@Autowired
protected ResourceHistoryCalculator myResourceHistoryCalculator;

@Autowired
private PlatformTransactionManager myTransactionManager;

@Autowired
protected ResourceHistoryCalculator myResourceHistoryCalculator;
private EntityManagerFactory myEntityManagerFactory;

protected TagDefinitionDao tagDefinitionDao;
protected final CodingSpy myCodingSpy = new CodingSpy();

@VisibleForTesting
Expand Down Expand Up @@ -484,7 +475,8 @@ protected TagDefinition getTagOrNull(

if (retVal == null) {
// actual DB hit(s) happen here
retVal = getOrCreateTag(theTagType, theScheme, theTerm, theLabel, theVersion, theUserSelected);
retVal = tagDefinitionDao.getOrCreateTag(
theTagType, theScheme, theTerm, theLabel, theVersion, theUserSelected);

TransactionSynchronization sync = new AddTagDefinitionToCacheAfterCommitSynchronization(key, retVal);
TransactionSynchronizationManager.registerSynchronization(sync);
Expand All @@ -496,124 +488,6 @@ protected TagDefinition getTagOrNull(
return retVal;
}

/**
* Gets the tag defined by the fed in values, or saves it if it does not
* exist.
* <p>
* Can also throw an InternalErrorException if something bad happens.
*/
private TagDefinition getOrCreateTag(
TagTypeEnum theTagType,
String theScheme,
String theTerm,
String theLabel,
String theVersion,
Boolean theUserSelected) {

TypedQuery<TagDefinition> q = buildTagQuery(theTagType, theScheme, theTerm, theVersion, theUserSelected);
q.setMaxResults(1);

TransactionTemplate template = new TransactionTemplate(myTransactionManager);
template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);

// this transaction will attempt to get or create the tag,
// repeating (on any failure) 10 times.
// if it fails more than this, we will throw exceptions
TagDefinition retVal;
int count = 0;
HashSet<Throwable> throwables = new HashSet<>();
do {
try {
retVal = template.execute(new TransactionCallback<TagDefinition>() {

// do the actual DB call(s) to read and/or write the values
private TagDefinition readOrCreate() {
TagDefinition val;
try {
val = q.getSingleResult();
} catch (NoResultException e) {
val = new TagDefinition(theTagType, theScheme, theTerm, theLabel);
val.setVersion(theVersion);
val.setUserSelected(theUserSelected);
myEntityManager.persist(val);
}
return val;
}

@Override
public TagDefinition doInTransaction(TransactionStatus status) {
TagDefinition tag = null;

try {
tag = readOrCreate();
} catch (Exception ex) {
// log any exceptions - just in case
// they may be signs of things to come...
ourLog.warn(
"Tag read/write failed: "
+ ex.getMessage() + ". "
+ "This is not a failure on its own, "
+ "but could be useful information in the result of an actual failure.",
ex);
throwables.add(ex);
}

return tag;
}
});
} catch (Exception ex) {
// transaction template can fail if connections to db are exhausted and/or timeout
ourLog.warn(
"Transaction failed with: {}. Transaction will rollback and be reattempted.", ex.getMessage());
retVal = null;
}
count++;
} while (retVal == null && count < TOTAL_TAG_READ_ATTEMPTS);

if (retVal == null) {
// if tag is still null,
// something bad must be happening
// - throw
String msg = throwables.stream().map(Throwable::getMessage).collect(Collectors.joining(", "));
throw new InternalErrorException(Msg.code(2023)
+ "Tag get/create failed after "
+ TOTAL_TAG_READ_ATTEMPTS
+ " attempts with error(s): "
+ msg);
}

return retVal;
}

private TypedQuery<TagDefinition> buildTagQuery(
TagTypeEnum theTagType, String theScheme, String theTerm, String theVersion, Boolean theUserSelected) {
CriteriaBuilder builder = myEntityManager.getCriteriaBuilder();
CriteriaQuery<TagDefinition> cq = builder.createQuery(TagDefinition.class);
Root<TagDefinition> from = cq.from(TagDefinition.class);

List<Predicate> predicates = new ArrayList<>();
predicates.add(builder.and(
builder.equal(from.get("myTagType"), theTagType), builder.equal(from.get("myCode"), theTerm)));

predicates.add(
isBlank(theScheme)
? builder.isNull(from.get("mySystem"))
: builder.equal(from.get("mySystem"), theScheme));

predicates.add(
isBlank(theVersion)
? builder.isNull(from.get("myVersion"))
: builder.equal(from.get("myVersion"), theVersion));

predicates.add(
isNull(theUserSelected)
? builder.isNull(from.get("myUserSelected"))
: builder.equal(from.get("myUserSelected"), theUserSelected));

cq.where(predicates.toArray(new Predicate[0]));
return myEntityManager.createQuery(cq);
}

void incrementId(T theResource, ResourceTable theSavedEntity, IIdType theResourceId) {
if (theResourceId == null || theResourceId.getVersionIdPart() == null) {
theSavedEntity.initializeVersion();
Expand Down Expand Up @@ -933,7 +807,7 @@ protected void postUpdate(ResourceTable theEntity, T theResource, RequestDetails
@Override
@CoverageIgnore
public BaseHasResource readEntity(IIdType theValueId, RequestDetails theRequest) {
throw new NotImplementedException(Msg.code(927) + "");
throw new NotImplementedException(Msg.code(927));
}

/**
Expand Down Expand Up @@ -1840,7 +1714,7 @@ protected void validateResourceForStorage(T theResource, ResourceTable theEntity

@PostConstruct
public void start() {
// nothing yet
this.tagDefinitionDao = new TagDefinitionDao(myEntityManagerFactory, myTransactionManager);
}

@VisibleForTesting
Expand Down
Loading

0 comments on commit 7534eb8

Please sign in to comment.