Add resource type to the HFJ_RESOURCE_TYPE primary key to avoid collisions (#7171)

* 7152 - Add resource type to HFJ_RESOURCE_MODIFIED's PK to avoid conflicts

* 7495 Fix integration tests affected by changes

* 7495 Add unit tests plus some refinements

* 7495: incorporate feedback from code review

* 7495 - Add a test to verify updates work as expected when resources share external ID and version

* 7152: Adjust version of DB we are migrating to

* 7152: Add entry in upgrade.md regarding database changes

Author: another-luis

hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/8_4_0/7152-add-resource-type-to-primary-key-to-guarantee-uniqueness.yaml

@@ -0,0 +1,6 @@
+---
+type: fix
+issue: 7152
+jira: SMILE-10226
+title: "Modified resources could potentially be removed from the subscription list of they shared ID and Version with 
+others. This has been fixed."

hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/8_4_0/upgrade.md

@@ -27,3 +27,6 @@ As of `8.3.12-SNAPSHOT`, HAPI-FHIR snapshots are now published on [Maven Central
 
 * `$export` and `$everything` operations on Patient Compartment (instance or type) will no longer return `List` or `Group` resources, regardless of auth rules.
 
+## Zero-Downtime Upgrades with Subscriptions
+
+The database upgrade includes changes to the `HFJ_RESOURCE_MODIFIED` table. This table holds transitional data about resources that have been modified. Under normal conditions, this table will be empty or nearly so. It is recommended to verify that subscriptions are running smoothly to avoid prolonged table locks while the table structure is updated.

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/migrate/tasks/HapiFhirJpaMigrationTasks.java

@@ -130,6 +130,15 @@ public HapiFhirJpaMigrationTasks(Set<String> theFlags) {
 		init780();
 		init820();
 		init840();
+		init860();
+	}
+
+	protected void init860() {
+		Builder version = forVersion(VersionEnum.V8_4_0);
+		{
+			version.onTable("HFJ_RESOURCE_MODIFIED").dropPrimaryKey("20250729.1");
+			version.onTable("HFJ_RESOURCE_MODIFIED").addPrimaryKey("20250729.2", "RES_ID", "RES_VER", "RESOURCE_TYPE");
+		}
 	}
 
 	protected void init840() {

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/ResourceModifiedMessagePersistenceSvcImpl.java

@@ -171,11 +171,11 @@ ResourceModifiedEntity createEntityFrom(ResourceModifiedMessage theMsg) {
 		IIdType theMsgId = theMsg.getPayloadId(myFhirContext);
 
 		ResourceModifiedEntity resourceModifiedEntity = new ResourceModifiedEntity();
-		resourceModifiedEntity.setResourceModifiedEntityPK(with(theMsgId.getIdPart(), theMsgId.getVersionIdPart()));
+		resourceModifiedEntity.setResourceModifiedEntityPK(
+				with(theMsgId.getIdPart(), theMsgId.getVersionIdPart(), theMsgId.getResourceType()));
 
 		String partialModifiedMessage = getPayloadLessMessageAsString(theMsg);
 		resourceModifiedEntity.setSummaryResourceModifiedMessage(partialModifiedMessage);
-		resourceModifiedEntity.setResourceType(theMsgId.getResourceType());
 		resourceModifiedEntity.setCreatedTime(new Date());
 
 		return resourceModifiedEntity;

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/entity/PersistedResourceModifiedMessageEntityPK.java

@@ -35,6 +35,9 @@ public class PersistedResourceModifiedMessageEntityPK implements IPersistedResou
 	@Column(name = "RES_VER", length = 8, nullable = false)
 	private String myResourceVersion;
 
+	@Column(name = "RESOURCE_TYPE", length = ResourceTable.RESTYPE_LEN, nullable = false)
+	private String myResourceType;
+
 	public String getResourcePid() {
 		return myResourcePid;
 	}
@@ -53,27 +56,40 @@ public PersistedResourceModifiedMessageEntityPK setResourceVersion(String theRes
 		return this;
 	}
 
-	public static PersistedResourceModifiedMessageEntityPK with(String theResourcePid, String theResourceVersion) {
+	public String getResourceType() {
+		return myResourceType;
+	}
+
+	public PersistedResourceModifiedMessageEntityPK setResourceType(String theResourceType) {
+		myResourceType = theResourceType;
+		return this;
+	}
+
+	public static PersistedResourceModifiedMessageEntityPK with(
+			String theResourcePid, String theResourceVersion, String theResourceType) {
 		return new PersistedResourceModifiedMessageEntityPK()
 				.setResourcePid(theResourcePid)
-				.setResourceVersion(theResourceVersion);
+				.setResourceVersion(theResourceVersion)
+				.setResourceType(theResourceType);
 	}
 
 	@Override
 	public boolean equals(Object theO) {
 		if (this == theO) return true;
 		if (theO == null || getClass() != theO.getClass()) return false;
 		PersistedResourceModifiedMessageEntityPK that = (PersistedResourceModifiedMessageEntityPK) theO;
-		return myResourcePid.equals(that.myResourcePid) && myResourceVersion.equals(that.myResourceVersion);
+		return myResourcePid.equals(that.myResourcePid)
+				&& myResourceVersion.equals(that.myResourceVersion)
+				&& myResourceType.equals(that.myResourceType);
 	}
 
 	@Override
 	public int hashCode() {
-		return Objects.hash(myResourcePid, myResourceVersion);
+		return Objects.hash(myResourcePid, myResourceVersion, myResourceType);
 	}
 
 	@Override
 	public String toString() {
-		return myResourcePid + "/" + myResourceVersion;
+		return myResourceType + "/" + myResourcePid + "/" + myResourceVersion;
 	}
 }

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/entity/ResourceModifiedEntity.java

@@ -53,9 +53,6 @@ public class ResourceModifiedEntity implements IPersistedResourceModifiedMessage
 	@Temporal(TemporalType.TIMESTAMP)
 	private Date myCreatedTime;
 
-	@Column(name = "RESOURCE_TYPE", length = ResourceTable.RESTYPE_LEN, nullable = false)
-	private String myResourceType;
-
 	public PersistedResourceModifiedMessageEntityPK getResourceModifiedEntityPK() {
 		return myResourceModifiedEntityPK;
 	}
@@ -68,12 +65,7 @@ public ResourceModifiedEntity setResourceModifiedEntityPK(
 
 	@Override
 	public String getResourceType() {
-		return myResourceType;
-	}
-
-	public ResourceModifiedEntity setResourceType(String theResourceType) {
-		myResourceType = theResourceType;
-		return this;
+		return myResourceModifiedEntityPK == null ? null : myResourceModifiedEntityPK.getResourceType();
 	}
 
 	@Override

hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/subscription/BaseSubscriptionsR4Test.java

@@ -21,6 +21,7 @@
 import ca.uhn.fhir.util.BundleUtil;
 import ca.uhn.test.concurrency.PointcutLatch;
 import com.apicatalog.jsonld.StringUtils;
+import com.google.common.base.Strings;
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.PostConstruct;
 import net.ttddyy.dsproxy.QueryCount;
@@ -224,21 +225,45 @@ protected Observation buildBaseObservation(String theCode, String theSystem) {
 	}
 
 	protected Patient sendPatient() {
+		return sendPatient(null);
+	}
+
+	protected Patient sendPatient(String thePatientId) {
 		Patient patient = new Patient();
-		patient.setActive(true);
+		if(!Strings.isNullOrEmpty(thePatientId)) {
+			patient.setId(thePatientId);
+		}
 
-		IIdType id = myPatientDao.create(patient).getId();
-		patient.setId(id);
+		DaoMethodOutcome outcome;
+		if(Strings.isNullOrEmpty(thePatientId)) {
+			outcome = myPatientDao.create(patient);
+			patient.setId( outcome.getId());
+		} else {
+			myPatientDao.update(patient);
+		}
 
+		patient.setActive(true);
 		return patient;
 	}
 
 	protected Organization sendOrganization() {
+		return sendOrganization(null);
+	}
+
+	protected Organization sendOrganization( String theOrgId) {
 		Organization org = new Organization();
-		org.setName("ORG");
+		org.setName("ORG" + (theOrgId == null ? "" : theOrgId));
+		if(!Strings.isNullOrEmpty(theOrgId)) {
+			org.setId(theOrgId);
+		}
 
-		IIdType id = myOrganizationDao.create(org).getId();
-		org.setId(id);
+		DaoMethodOutcome outcome;
+		if(Strings.isNullOrEmpty(theOrgId)) {
+			outcome = myOrganizationDao.create(org);
+			org.setId(outcome.getId());
+		} else {
+			myOrganizationDao.update(org);
+		}
 
 		return org;
 	}

hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/subscription/async/AsyncSubscriptionMessageSubmissionIT.java

@@ -100,9 +100,9 @@ public void runDeliveryPass_withManyResources_isBatchedAndKeepsResourceUsageDown
 		int numberOfResourcesToCreate = factor * AsyncResourceModifiedSubmitterSvc.MAX_LIMIT;
 
 		ResourceModifiedEntity entity = new ResourceModifiedEntity();
-		entity.setResourceType(resourceType);
 		PersistedResourceModifiedMessageEntityPK rpm = new PersistedResourceModifiedMessageEntityPK();
 		rpm.setResourceVersion("1");
+		rpm.setResourceType(resourceType);
 		entity.setResourceModifiedEntityPK(rpm);
 
 		// we reuse the same exact msg content to avoid

hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/subscription/message/MessageSubscriptionR4Test.java

@@ -26,6 +26,7 @@
 import org.hl7.fhir.r4.model.Observation;
 import org.hl7.fhir.r4.model.Organization;
 import org.hl7.fhir.r4.model.Patient;
+import org.hl7.fhir.r4.model.ResourceType;
 import org.hl7.fhir.r4.model.Subscription;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -238,13 +239,61 @@ public void testMethodDeleteByPK_whenEntityExists_willDeleteTheEntityAndReturnTr
 			.hasSize(0);
 	}
 
+	@Test
+	public void testMethodDeleteByPK_whenEntitiesExistWithRepeatedId_willDeleteTheCorrectEntityAndReturnTrue(){
+		String commonId = "common-id";
+
+		mySubscriptionTestUtil.unregisterSubscriptionInterceptor();
+
+		// given
+		TransactionTemplate transactionTemplate = new TransactionTemplate(myTxManager);
+
+		Patient patient = sendPatient(commonId);
+		ResourceModifiedMessage patientResourceModifiedMessage = new ResourceModifiedMessage(myFhirContext, patient, BaseResourceMessage.OperationTypeEnum.CREATE);
+		IPersistedResourceModifiedMessage persistedPatientResourceModifiedMessage = myResourceModifiedMessagePersistenceSvc.persist(patientResourceModifiedMessage);
+
+		Organization org = sendOrganization(commonId);
+		ResourceModifiedMessage orgResourceModifiedMessage = new ResourceModifiedMessage(myFhirContext, org, BaseResourceMessage.OperationTypeEnum.CREATE);
+		myResourceModifiedMessagePersistenceSvc.persist(orgResourceModifiedMessage);
+
+		// when
+		boolean wasDeleted = transactionTemplate.execute(tx -> myResourceModifiedMessagePersistenceSvc.deleteByPK(persistedPatientResourceModifiedMessage.getPersistedResourceModifiedMessagePk()));
+
+		// then
+		assertTrue(wasDeleted);
+		Page<IPersistedResourceModifiedMessage> messages = myResourceModifiedMessagePersistenceSvc.findAllOrderedByCreatedTime(Pageable.unpaged());
+		assertThat(messages).hasSize(1);
+		assertEquals(ResourceType.Organization.name(), messages.stream().toList().get(0).getResourceType());
+
+	}
+
+	@Test
+	public void testMethodPersist_AddEntriesWithSameExternalIdAndVersion_expectSuccess(){
+		mySubscriptionTestUtil.unregisterSubscriptionInterceptor();
+
+		// given
+		String commonId = "common-id";
+		Patient patient = sendPatient(commonId);
+		Organization organization = sendOrganization(commonId);
+
+		ResourceModifiedMessage patientResourceModifiedMessage = new ResourceModifiedMessage(myFhirContext, patient, BaseResourceMessage.OperationTypeEnum.CREATE);
+		ResourceModifiedMessage organizationResourceModifiedMessage = new ResourceModifiedMessage(myFhirContext, organization, BaseResourceMessage.OperationTypeEnum.CREATE);
+
+		// when
+		myResourceModifiedMessagePersistenceSvc.persist(patientResourceModifiedMessage);
+		myResourceModifiedMessagePersistenceSvc.persist(organizationResourceModifiedMessage);
+
+		// then
+		assertEquals(2,  myResourceModifiedMessagePersistenceSvc.getMessagePersistedCount());
+	}
+
 	@Test
 	public void testMethodDeleteByPK_whenEntityDoesNotExist_willReturnFalse(){
 		mySubscriptionTestUtil.unregisterSubscriptionInterceptor();
 
 		// given
 		TransactionTemplate transactionTemplate = new TransactionTemplate(myTxManager);
-		IPersistedResourceModifiedMessagePK nonExistentResourceWithPk = PersistedResourceModifiedMessageEntityPK.with("one", "one");
+		IPersistedResourceModifiedMessagePK nonExistentResourceWithPk = PersistedResourceModifiedMessageEntityPK.with("one", "one", ResourceType.Patient.toString());
 
 		// when
 		boolean wasDeleted = transactionTemplate.execute(tx -> myResourceModifiedMessagePersistenceSvc.deleteByPK(nonExistentResourceWithPk));