added fix for subscription in default partition when default partitio… (#3670)

* added fix for subscription in default partition when default partition's id is null

* added changelog

Co-authored-by: Long Ma <long@smilecdr.com>

hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/6_1_0/3669-multitenant-subscription-default-partition-fix.yaml

@@ -0,0 +1,6 @@
+---
+type: fix
+issue: 2669
+title: "Previously subscriptions in a partition with the id null will be matched against incoming resources from all partitions.
+Changed to subscriptions will only match against incoming resources in the partition the subscription exists in unless
+cross partition subscription is enabled and the subscription has the appropriate extension."

hapi-fhir-jpaserver-subscription/src/main/java/ca/uhn/fhir/jpa/subscription/match/matcher/subscriber/SubscriptionMatchingSubscriber.java

@@ -127,7 +127,7 @@ public class SubscriptionMatchingSubscriber implements MessageHandler {
 		for (ActiveSubscription nextActiveSubscription : subscriptions) {
 			// skip if the partitions don't match
 			CanonicalSubscription subscription = nextActiveSubscription.getSubscription();
-			if (subscription != null && subscription.getRequestPartitionId() != null && theMsg.getPartitionId() != null &&
+			if (subscription != null && theMsg.getPartitionId() != null &&
 				theMsg.getPartitionId().hasPartitionIds() && !subscription.getCrossPartitionEnabled() &&
 				!theMsg.getPartitionId().hasPartitionId(subscription.getRequestPartitionId())) {
 				continue;

hapi-fhir-jpaserver-subscription/src/test/java/ca/uhn/fhir/jpa/subscription/module/subscriber/SubscriptionMatchingSubscriberTest.java

@@ -245,6 +245,42 @@ public class SubscriptionMatchingSubscriberTest extends BaseBlockingQueueSubscri
 		mySubscriptionResourceNotMatched.awaitExpected();
 	}
 
+	@Test
+	public void testSubscriptionOnDefaultPartitionAndResourceOnDiffPartitionNotMatch() throws InterruptedException {
+		myPartitionSettings.setPartitioningEnabled(true);
+		String payload = "application/fhir+json";
+
+		String code = "1000000050";
+		String criteria = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+
+		RequestPartitionId requestPartitionId = RequestPartitionId.defaultPartition();
+		Subscription subscription = makeActiveSubscription(criteria, payload, ourListenerServerBase);
+		mockSubscriptionRead(requestPartitionId, subscription);
+		sendSubscription(subscription, requestPartitionId, true);
+
+		mySubscriptionResourceNotMatched.setExpectedCount(1);
+		sendObservation(code, "SNOMED-CT", RequestPartitionId.fromPartitionId(1));
+		mySubscriptionResourceNotMatched.awaitExpected();
+	}
+
+	@Test
+	public void testSubscriptionOnAPartitionAndResourceOnDefaultPartitionNotMatch() throws InterruptedException {
+		myPartitionSettings.setPartitioningEnabled(true);
+		String payload = "application/fhir+json";
+
+		String code = "1000000050";
+		String criteria = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+
+		RequestPartitionId requestPartitionId = RequestPartitionId.fromPartitionId(1);
+		Subscription subscription = makeActiveSubscription(criteria, payload, ourListenerServerBase);
+		mockSubscriptionRead(requestPartitionId, subscription);
+		sendSubscription(subscription, requestPartitionId, true);
+
+		mySubscriptionResourceNotMatched.setExpectedCount(1);
+		sendObservation(code, "SNOMED-CT", RequestPartitionId.defaultPartition());
+		mySubscriptionResourceNotMatched.awaitExpected();
+	}
+
 	@Test
 	public void testSubscriptionOnOnePartitionMatchResourceOnMultiplePartitions() throws InterruptedException {
 		myPartitionSettings.setPartitioningEnabled(true);