Skip to content

Commit 69fede9

Browse files
XcrigXXcrigX
authored
add property to turn on userRequestRetryVersionConflictsInterceptor (#752)
1 parent 5d84d9f commit 69fede9

File tree

4 files changed

+181
-3
lines changed

4 files changed

+181
-3
lines changed

‎src/main/java/ca/uhn/fhir/jpa/starter/AppProperties.java

+10-3
Original file line numberDiff line numberDiff line change
@@ -106,8 +106,9 @@ public class AppProperties {
106106
private final List<String> custom_provider_classes = new ArrayList<>();
107107
private Boolean upliftedRefchains_enabled = false;
108108

109-
private List<Integer> search_prefetch_thresholds = new ArrayList<>();
109+
private boolean userRequestRetryVersionConflictsInterceptorEnabled = false;
110110

111+
private List<Integer> search_prefetch_thresholds = new ArrayList<>();
111112

112113
public List<String> getCustomInterceptorClasses() {
113114
return custom_interceptor_classes;
@@ -671,6 +672,14 @@ public void setUpliftedRefchains_enabled(boolean upliftedRefchains_enabled) {
671672
this.upliftedRefchains_enabled = upliftedRefchains_enabled;
672673
}
673674

675+
public Boolean getUserRequestRetryVersionConflictsInterceptorEnabled() {
676+
return userRequestRetryVersionConflictsInterceptorEnabled;
677+
}
678+
679+
public void setUserRequestRetryVersionConflictsInterceptorEnabled(Boolean userRequestRetryVersionConflictsInterceptorEnabled) {
680+
this.userRequestRetryVersionConflictsInterceptorEnabled = userRequestRetryVersionConflictsInterceptorEnabled;
681+
}
682+
674683
public static class Cors {
675684
private Boolean allow_Credentials = true;
676685
private List<String> allowed_origin = List.of("*");
@@ -690,8 +699,6 @@ public Boolean getAllow_Credentials() {
690699
public void setAllow_Credentials(Boolean allow_Credentials) {
691700
this.allow_Credentials = allow_Credentials;
692701
}
693-
694-
695702
}
696703

697704
public static class Logger {

‎src/main/java/ca/uhn/fhir/jpa/starter/common/StarterJpaConfig.java

+5
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import ca.uhn.fhir.jpa.delete.ThreadSafeResourceDeleterSvc;
2929
import ca.uhn.fhir.jpa.graphql.GraphQLProvider;
3030
import ca.uhn.fhir.jpa.interceptor.CascadingDeleteInterceptor;
31+
import ca.uhn.fhir.jpa.interceptor.UserRequestRetryVersionConflictsInterceptor;
3132
import ca.uhn.fhir.jpa.interceptor.validation.RepositoryValidatingInterceptor;
3233
import ca.uhn.fhir.jpa.ips.provider.IpsOperationProvider;
3334
import ca.uhn.fhir.jpa.model.config.SubscriptionSettings;
@@ -457,6 +458,10 @@ public RestfulServer restfulServer(
457458
fhirServer.registerProvider(theIpsOperationProvider.get());
458459
}
459460

461+
if (appProperties.getUserRequestRetryVersionConflictsInterceptorEnabled() ) {
462+
fhirServer.registerInterceptor(new UserRequestRetryVersionConflictsInterceptor());
463+
}
464+
460465
// register custom providers
461466
registerCustomProviders(fhirServer, appContext, appProperties.getCustomProviderClasses());
462467

‎src/main/resources/application.yaml

+2
Original file line numberDiff line numberDiff line change
@@ -214,6 +214,8 @@ hapi:
214214
narrative_enabled: false
215215
mdm_enabled: false
216216
mdm_rules_json_location: "mdm-rules.json"
217+
## see: https://hapifhir.io/hapi-fhir/docs/interceptors/built_in_server_interceptors.html#jpa-server-retry-on-version-conflicts
218+
# userRequestRetryVersionConflictsInterceptorEnabled : false
217219
# local_base_urls:
218220
# - https://hapi.fhir.org/baseR4
219221
logical_urls:

‎src/test/java/ca/uhn/fhir/jpa/starter/ParallelUpdatesVersionConflictTest.java

+164
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,164 @@
1+
package ca.uhn.fhir.jpa.starter;
2+
3+
import java.util.ArrayList;
4+
import java.util.List;
5+
import java.util.concurrent.Callable;
6+
import java.util.concurrent.ExecutionException;
7+
import java.util.concurrent.ExecutorService;
8+
import java.util.concurrent.Executors;
9+
import java.util.concurrent.Future;
10+
11+
import ca.uhn.fhir.rest.server.exceptions.ResourceVersionConflictException;
12+
import org.hl7.fhir.r4.model.Bundle.BundleEntryComponent;
13+
import org.hl7.fhir.r4.model.Patient;
14+
import org.hl7.fhir.r4.model.Bundle;
15+
import org.hl7.fhir.r4.model.Bundle.BundleType;
16+
import org.hl7.fhir.r4.model.Bundle.HTTPVerb;
17+
import org.junit.jupiter.api.Assertions;
18+
import org.junit.jupiter.api.BeforeEach;
19+
import org.junit.jupiter.api.Test;
20+
import org.springframework.beans.factory.annotation.Autowired;
21+
import org.springframework.boot.test.context.SpringBootTest;
22+
import org.springframework.boot.test.web.server.LocalServerPort;
23+
24+
import ca.uhn.fhir.context.FhirContext;
25+
import ca.uhn.fhir.jpa.api.dao.IFhirResourceDao;
26+
import ca.uhn.fhir.rest.api.MethodOutcome;
27+
import ca.uhn.fhir.rest.client.api.IGenericClient;
28+
import ca.uhn.fhir.rest.client.api.ServerValidationModeEnum;
29+
30+
import static org.junit.jupiter.api.Assertions.assertThrows;
31+
32+
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, classes = {Application.class}, properties = {
33+
"spring.datasource.url=jdbc:h2:mem:dbr4",
34+
"hapi.fhir.fhir_version=r4",
35+
"hapi.fhir.userRequestRetryVersionConflictsInterceptorEnabled=true"
36+
})
37+
38+
/**
39+
* This class tests running parallel updates to a single resource with and without setting the 'X-Retry-On-Version-Conflict' header
40+
* to ensure we get the expected behavior of detecting conflicts
41+
*/
42+
public class ParallelUpdatesVersionConflictTest {
43+
44+
private static final org.slf4j.Logger ourLog = org.slf4j.LoggerFactory.getLogger(ParallelUpdatesVersionConflictTest.class);
45+
46+
@LocalServerPort
47+
private int port;
48+
49+
private IGenericClient client;
50+
private FhirContext ctx;
51+
52+
@BeforeEach
53+
void setUp() {
54+
ctx = FhirContext.forR4();
55+
ctx.getRestfulClientFactory().setServerValidationMode(ServerValidationModeEnum.NEVER);
56+
ctx.getRestfulClientFactory().setSocketTimeout(1200 * 1000);
57+
String ourServerBase = "http://localhost:" + port + "/fhir/";
58+
client = ctx.newRestfulGenericClient(ourServerBase);
59+
}
60+
61+
@Test
62+
void testParallelResourceUpdateBundle() throws Throwable {
63+
//send 10 bundles with updates to the patient in parallel, except the header to deconflict them
64+
Patient pat = new Patient();
65+
String patId = client.create().resource(pat).execute().getId().getIdPart();
66+
launchThreads(patId, true, "X-Retry-On-Version-Conflict");
67+
}
68+
69+
@Test
70+
void testParallelResourceUpdateNoBundle() throws Throwable {
71+
//send 10 resource puts to the patient in parallel, except the header to deconflict them
72+
Patient pat = new Patient();
73+
String patId = client.create().resource(pat).execute().getId().getIdPart();
74+
launchThreads(patId, false, "X-Retry-On-Version-Conflict");
75+
}
76+
77+
@Test
78+
void testParallelResourceUpdateBundleExpectConflict() {
79+
//send 10 bundles with updates to the patient in parallel, expect a ResourceVersionConflictException since we are not setting the retry header
80+
Patient pat = new Patient();
81+
String patId = client.create().resource(pat).execute().getId().getIdPart();
82+
ResourceVersionConflictException exception = assertThrows(ResourceVersionConflictException.class, () ->
83+
launchThreads(patId, true, "someotherheader"));
84+
}
85+
86+
@Test
87+
void testParallelResourceUpdateNoBundleExpectConflict() {
88+
//send 10 resource puts to the patient in parallel, expect a ResourceVersionConflictException since we are not setting the retry header
89+
Patient pat = new Patient();
90+
String patId = client.create().resource(pat).execute().getId().getIdPart();
91+
ResourceVersionConflictException exception = assertThrows(ResourceVersionConflictException.class, () ->
92+
launchThreads(patId, false, "someotherheader"));
93+
}
94+
95+
private void launchThreads(String patientId, boolean useBundles, String headerName) throws Throwable {
96+
int threadCnt = 10;
97+
ExecutorService execSvc = Executors.newFixedThreadPool(threadCnt);
98+
99+
//launch a bunch of threads at the same time that update the same patient
100+
List<Callable<Integer>> callables = new ArrayList<>();
101+
for (int i = 0; i < threadCnt; i++) {
102+
final int cnt = i;
103+
Callable<Integer> callable = new Callable<>() {
104+
@Override
105+
public Integer call() throws Exception {
106+
Patient pat = new Patient();
107+
//make sure to change something so the server doesnt short circuit on a no-op
108+
pat.addName().setFamily("fam-" + cnt);
109+
pat.setId(patientId);
110+
111+
if( useBundles) {
112+
Bundle b = new Bundle();
113+
b.setType(BundleType.TRANSACTION);
114+
BundleEntryComponent bec = b.addEntry();
115+
bec.setResource(pat);
116+
//bec.setFullUrl("Patient/" + patId);
117+
Bundle.BundleEntryRequestComponent req = bec.getRequest();
118+
req.setUrl("Patient/" + patientId);
119+
req.setMethod(HTTPVerb.PUT);
120+
bec.setRequest(req);
121+
122+
Bundle returnBundle = client.transaction().withBundle(b)
123+
.withAdditionalHeader(headerName, "retry; max-retries=10")
124+
.execute();
125+
126+
String statusString = returnBundle.getEntryFirstRep().getResponse().getStatus();
127+
ourLog.trace("statusString->{}", statusString);
128+
try {
129+
return Integer.parseInt(statusString.substring(0,3));
130+
}catch(NumberFormatException nfe) {
131+
return 500;
132+
}
133+
}
134+
else {
135+
MethodOutcome outcome = client.update().resource(pat).withId(patientId)
136+
.withAdditionalHeader(headerName, "retry; max-retries=10")
137+
.execute();
138+
ourLog.trace("updated patient: " + outcome.getResponseStatusCode());
139+
return outcome.getResponseStatusCode();
140+
}
141+
}
142+
};
143+
callables.add(callable);
144+
}
145+
146+
List<Future<Integer>> futures = new ArrayList<>();
147+
148+
//launch them all at once
149+
for (Callable<Integer> callable : callables) {
150+
futures.add(execSvc.submit(callable));
151+
}
152+
153+
//wait for calls to complete
154+
for (Future<Integer> future : futures) {
155+
try {
156+
Integer httpResponseCode = future.get();
157+
Assertions.assertEquals(200, httpResponseCode);
158+
} catch (InterruptedException | ExecutionException e) {
159+
//throw the ResourceVersionConflictException back up so we can test it
160+
throw e.getCause();
161+
}
162+
}
163+
}
164+
}

0 commit comments

Comments
 (0)