Merge remote-tracking branch 'origin/master' into do-20231213-core-bump-6-2-6

hapi-fhir-base/src/main/java/ca/uhn/fhir/interceptor/model/RequestPartitionId.java

@@ -38,6 +38,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 import static org.apache.commons.lang3.ObjectUtils.defaultIfNull;
@@ -134,6 +135,9 @@ public class RequestPartitionId implements IModelJson {
 		if (hasPartitionNames()) {
 			b.append("names", getPartitionNames());
 		}
+		if (myAllPartitions) {
+			b.append("allPartitions", myAllPartitions);
+		}
 		return b.build();
 	}
 
@@ -216,7 +220,7 @@ public class RequestPartitionId implements IModelJson {
 	}
 
 	public List<Integer> getPartitionIdsWithoutDefault() {
-		return getPartitionIds().stream().filter(t -> t != null).collect(Collectors.toList());
+		return getPartitionIds().stream().filter(Objects::nonNull).collect(Collectors.toList());
 	}
 
 	@Nullable

hapi-fhir-base/src/main/java/ca/uhn/fhir/util/TaskChunker.java

@@ -20,10 +20,13 @@ package ca.uhn.fhir.util;
  * #L%
  */
 
+import jakarta.annotation.Nonnull;
+
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.function.Consumer;
+import java.util.stream.Stream;
 
 /**
  * This utility takes an input collection, breaks it up into chunks of a
@@ -49,4 +52,9 @@ public class TaskChunker<T> {
 			theBatchConsumer.accept(batch);
 		}
 	}
+
+	@Nonnull
+	public <T> Stream<List<T>> chunk(Stream<T> theStream, int theChunkSize) {
+		return StreamUtil.partition(theStream, theChunkSize);
+	}
 }

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/BaseHapiFhirResourceDao.java

@@ -1540,7 +1540,7 @@ public abstract class BaseHapiFhirResourceDao<T extends IBaseResource> extends B
 				.withRequest(theRequest)
 				.withTransactionDetails(transactionDetails)
 				.withRequestPartitionId(requestPartitionId)
-				.execute(() -> doReadInTransaction(theId, theRequest, theDeletedOk, requestPartitionId));
+				.read(() -> doReadInTransaction(theId, theRequest, theDeletedOk, requestPartitionId));
 	}
 
 	private T doReadInTransaction(

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/index/IdHelperService.java

@@ -158,7 +158,8 @@ public class IdHelperService implements IIdHelperService<JpaPid> {
 			String theResourceId,
 			boolean theExcludeDeleted)
 			throws ResourceNotFoundException {
-		assert myDontCheckActiveTransactionForUnitTest || TransactionSynchronizationManager.isSynchronizationActive();
+		assert myDontCheckActiveTransactionForUnitTest || TransactionSynchronizationManager.isSynchronizationActive()
+				: "no transaction active";
 
 		if (theResourceId.contains("/")) {
 			theResourceId = theResourceId.substring(theResourceId.indexOf("/") + 1);

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/entity/PartitionEntity.java

@@ -26,6 +26,9 @@ import jakarta.persistence.Id;
 import jakarta.persistence.Table;
 import jakarta.persistence.UniqueConstraint;
 
+import java.util.ArrayList;
+import java.util.List;
+
 @Entity
 @Table(
 		name = "HFJ_PARTITION",
@@ -82,4 +85,20 @@ public class PartitionEntity {
 	public RequestPartitionId toRequestPartitionId() {
 		return RequestPartitionId.fromPartitionIdAndName(getId(), getName());
 	}
+
+	/**
+	 * Build a RequestPartitionId from the ids and names in the entities.
+	 * @param thePartitions the entities to use for ids and names
+	 * @return a single RequestPartitionId covering all the entities
+	 */
+	public static RequestPartitionId buildRequestPartitionId(List<PartitionEntity> thePartitions) {
+		List<Integer> ids = new ArrayList<>(thePartitions.size());
+		List<String> names = new ArrayList<>(thePartitions.size());
+		for (PartitionEntity nextPartition : thePartitions) {
+			ids.add(nextPartition.getId());
+			names.add(nextPartition.getName());
+		}
+
+		return RequestPartitionId.forPartitionIdsAndNames(names, ids, null);
+	}
 }

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/util/QueryChunker.java

@@ -25,6 +25,7 @@ import ca.uhn.fhir.util.TaskChunker;
 import java.util.Collection;
 import java.util.List;
 import java.util.function.Consumer;
+import java.util.stream.Stream;
 
 /**
  * As always, Oracle can't handle things that other databases don't mind.. In this
@@ -37,4 +38,8 @@ public class QueryChunker<T> extends TaskChunker<T> {
 	public void chunk(Collection<T> theInput, Consumer<List<T>> theBatchConsumer) {
 		chunk(theInput, SearchBuilder.getMaximumPageSize(), theBatchConsumer);
 	}
+
+	public Stream<List<T>> chunk(Stream<T> theStream) {
+		return chunk(theStream, SearchBuilder.getMaximumPageSize());
+	}
 }

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/entity/PartitionEntityTest.java

@@ -0,0 +1,31 @@
+package ca.uhn.fhir.jpa.entity;
+
+import ca.uhn.fhir.interceptor.model.RequestPartitionId;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
+
+class PartitionEntityTest {
+
+	@Test
+	void buildRequestPartitionFromList() {
+	    // given
+		List<PartitionEntity> entities = List.of(
+			new PartitionEntity().setId(1).setName("p1"),
+			new PartitionEntity().setId(2).setName("p2")
+		);
+
+	    // when
+		RequestPartitionId requestPartitionId = PartitionEntity.buildRequestPartitionId(entities);
+
+		// then
+	    assertThat(requestPartitionId, notNullValue());
+	    assertThat(requestPartitionId.getPartitionIds(), equalTo(List.of(1,2)));
+	    assertThat(requestPartitionId.getPartitionNames(), equalTo(List.of("p1","p2")));
+	}
+
+}

hapi-fhir-jpaserver-subscription/src/main/java/ca/uhn/fhir/jpa/subscription/match/registry/SubscriptionLoader.java

@@ -24,6 +24,7 @@ import ca.uhn.fhir.jpa.searchparam.SearchParameterMap;
 import ca.uhn.fhir.jpa.subscription.match.matcher.subscriber.SubscriptionActivatingSubscriber;
 import ca.uhn.fhir.rest.param.TokenOrListParam;
 import ca.uhn.fhir.rest.param.TokenParam;
+import ca.uhn.fhir.rest.server.util.ISearchParamRegistry;
 import ca.uhn.fhir.subscription.SubscriptionConstants;
 import jakarta.annotation.Nonnull;
 import org.apache.commons.lang3.StringUtils;
@@ -49,6 +50,9 @@ public class SubscriptionLoader extends BaseResourceCacheSynchronizer {
 	@Autowired
 	private SubscriptionCanonicalizer mySubscriptionCanonicalizer;
 
+	@Autowired
+	protected ISearchParamRegistry mySearchParamRegistry;
+
 	/**
 	 * Constructor
 	 */
@@ -157,11 +161,11 @@ public class SubscriptionLoader extends BaseResourceCacheSynchronizer {
 		} else {
 			error = "";
 		}
-		ourLog.error("Subscription "
+		ourLog.error(
-				+ theSubscription.getIdElement().getIdPart()
+				"Subscription {} could not be activated."
-				+ " could not be activated."
+						+ " This will not prevent startup, but it could lead to undesirable outcomes! {}",
-				+ " This will not prevent startup, but it could lead to undesirable outcomes! "
+				theSubscription.getIdElement().getIdPart(),
-				+ (StringUtils.isBlank(error) ? "" : "Error: " + error));
+				(StringUtils.isBlank(error) ? "" : "Error: " + error));
 	}
 
 	public void syncSubscriptions() {

hapi-fhir-jpaserver-subscription/src/main/java/ca/uhn/fhir/jpa/topic/SubscriptionTopicLoader.java

@@ -25,6 +25,7 @@ import ca.uhn.fhir.context.FhirVersionEnum;
 import ca.uhn.fhir.i18n.Msg;
 import ca.uhn.fhir.jpa.searchparam.SearchParameterMap;
 import ca.uhn.fhir.rest.param.TokenParam;
+import ca.uhn.fhir.rest.server.util.ISearchParamRegistry;
 import ca.uhn.fhir.subscription.SubscriptionConstants;
 import ca.uhn.fhir.util.Logs;
 import jakarta.annotation.Nonnull;
@@ -47,6 +48,9 @@ public class SubscriptionTopicLoader extends BaseResourceCacheSynchronizer {
 	@Autowired
 	private SubscriptionTopicRegistry mySubscriptionTopicRegistry;
 
+	@Autowired
+	protected ISearchParamRegistry mySearchParamRegistry;
+
 	/**
 	 * Constructor
 	 */

hapi-fhir-jpaserver-subscription/src/test/java/ca/uhn/fhir/jpa/subscription/match/registry/SubscriptionLoaderTest.java

@@ -13,27 +13,24 @@ import ca.uhn.fhir.rest.api.server.SystemRequestDetails;
 import ca.uhn.fhir.rest.server.SimpleBundleProvider;
 import ca.uhn.fhir.rest.server.util.ISearchParamRegistry;
 import ca.uhn.fhir.subscription.SubscriptionConstants;
+import ca.uhn.test.util.LogbackCaptureTestExtension;
 import ch.qos.logback.classic.Level;
-import ch.qos.logback.classic.Logger;
-import ch.qos.logback.classic.spi.ILoggingEvent;
-import ch.qos.logback.core.Appender;
 import org.hl7.fhir.instance.model.api.IBaseResource;
 import org.hl7.fhir.r4.model.Subscription;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
+import org.junit.jupiter.api.extension.RegisterExtension;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.Spy;
 import org.mockito.junit.jupiter.MockitoExtension;
-import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
 import java.util.List;
 
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItem;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -43,13 +40,11 @@ import static org.mockito.Mockito.when;
 @ExtendWith(MockitoExtension.class)
 public class SubscriptionLoaderTest {
 
-	private Logger ourLogger;
-
 	@Spy
 	private FhirContext myFhirContext = FhirContext.forR4Cached();
 
-	@Mock
+	@RegisterExtension
-	private Appender<ILoggingEvent> myAppender;
+	LogbackCaptureTestExtension myLogCapture = new LogbackCaptureTestExtension(SubscriptionLoader.class);
 
 	@Mock
 	private SubscriptionRegistry mySubscriptionRegistry;
@@ -81,15 +76,8 @@ public class SubscriptionLoaderTest {
 	@InjectMocks
 	private SubscriptionLoader mySubscriptionLoader;
 
-	private Level myStoredLogLevel;
-
 	@BeforeEach
 	public void init() {
-		ourLogger = (Logger) LoggerFactory.getLogger(SubscriptionLoader.class);
-
-		myStoredLogLevel = ourLogger.getLevel();
-		ourLogger.addAppender(myAppender);
-
 		when(myResourceChangeListenerRegistry.registerResourceResourceChangeListener(
 			anyString(),
 			any(SearchParameterMap.class),
@@ -102,12 +90,6 @@ public class SubscriptionLoaderTest {
 		mySubscriptionLoader.registerListener();
 	}
 
-	@AfterEach
-	public void end() {
-		ourLogger.detachAppender(myAppender);
-		ourLogger.setLevel(myStoredLogLevel);
-	}
-
 	private IBundleProvider getSubscriptionList(List<IBaseResource> theReturnedResource) {
 		IBundleProvider subscriptionList = new SimpleBundleProvider(theReturnedResource);
 
@@ -122,8 +104,6 @@ public class SubscriptionLoaderTest {
 		subscription.setId("Subscription/123");
 		subscription.setError("THIS IS AN ERROR");
 
-		ourLogger.setLevel(Level.ERROR);
-
 		// when
 		when(myDaoRegistry.getResourceDao("Subscription"))
 			.thenReturn(mySubscriptionDao);
@@ -149,15 +129,10 @@ public class SubscriptionLoaderTest {
 		verify(mySubscriptionDao)
 			.search(any(SearchParameterMap.class), any(SystemRequestDetails.class));
 
-		ArgumentCaptor<ILoggingEvent> eventCaptor = ArgumentCaptor.forClass(ILoggingEvent.class);
+		String expected = "Subscription "
-		verify(myAppender).doAppend(eventCaptor.capture());
-		String actual = "Subscription "
 			+ subscription.getIdElement().getIdPart()
 			+ " could not be activated.";
-		String msg = eventCaptor.getValue().getMessage();
+		assertThat(myLogCapture.getLogEvents(), hasItem(LogbackCaptureTestExtension.eventWithLevelAndMessageContains(Level.ERROR, expected)));
-		Assertions.assertTrue(msg
+		assertThat(myLogCapture.getLogEvents(), hasItem(LogbackCaptureTestExtension.eventWithMessageContains(subscription.getError())));
-			.contains(actual),
-			String.format("Expected %s, actual %s", msg, actual));
-		Assertions.assertTrue(msg.contains(subscription.getError()));
 	}
 }

hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/dao/tx/ReindexStepTest.java

@@ -44,6 +44,7 @@ public class ReindexStepTest {
 		ReindexJobParameters reindexJobParameters = new ReindexJobParameters();
 		reindexJobParameters.setRequestPartitionId(RequestPartitionId.fromPartitionId(expectedPartitionId));
 		when(myHapiTransactionService.withRequest(any())).thenCallRealMethod();
+		when(myHapiTransactionService.buildExecutionBuilder(any())).thenCallRealMethod();
 
 		// when
 		myReindexStep.doReindex(data, myDataSink, "index-id", "chunk-id", reindexJobParameters);

hapi-fhir-storage/src/main/java/ca/uhn/fhir/cache/BaseResourceCacheSynchronizer.java

@@ -29,7 +29,6 @@ import ca.uhn.fhir.jpa.searchparam.SearchParameterMap;
 import ca.uhn.fhir.jpa.searchparam.retry.Retrier;
 import ca.uhn.fhir.rest.api.server.IBundleProvider;
 import ca.uhn.fhir.rest.api.server.SystemRequestDetails;
-import ca.uhn.fhir.rest.server.util.ISearchParamRegistry;
 import ca.uhn.fhir.subscription.SubscriptionConstants;
 import com.google.common.annotations.VisibleForTesting;
 import jakarta.annotation.Nonnull;
@@ -56,9 +55,6 @@ public abstract class BaseResourceCacheSynchronizer implements IResourceChangeLi
 	public static final long REFRESH_INTERVAL = DateUtils.MILLIS_PER_MINUTE;
 	private final String myResourceName;
 
-	@Autowired
-	protected ISearchParamRegistry mySearchParamRegistry;
-
 	@Autowired
 	private IResourceChangeListenerRegistry myResourceChangeListenerRegistry;
 
@@ -71,10 +67,19 @@ public abstract class BaseResourceCacheSynchronizer implements IResourceChangeLi
 	private final Semaphore mySyncResourcesSemaphore = new Semaphore(1);
 	private final Object mySyncResourcesLock = new Object();
 
-	public BaseResourceCacheSynchronizer(String theResourceName) {
+	protected BaseResourceCacheSynchronizer(String theResourceName) {
 		myResourceName = theResourceName;
 	}
 
+	protected BaseResourceCacheSynchronizer(
+			String theResourceName,
+			IResourceChangeListenerRegistry theResourceChangeListenerRegistry,
+			DaoRegistry theDaoRegistry) {
+		myResourceName = theResourceName;
+		myDaoRegistry = theDaoRegistry;
+		myResourceChangeListenerRegistry = theResourceChangeListenerRegistry;
+	}
+
 	@PostConstruct
 	public void registerListener() {
 		if (myDaoRegistry.getResourceDaoOrNull(myResourceName) == null) {

hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/dao/IFhirResourceDao.java

@@ -374,7 +374,7 @@ public interface IFhirResourceDao<T extends IBaseResource> extends IDao {
 	 * The Stream MUST be closed to avoid leaking resources.
 	 * @param theParams the search
 	 * @param theRequest for partition target info
-	 * @return a Stream than MUST only be used within the calling transaction.
+	 * @return a Stream that MUST only be used within the calling transaction.
 	 */
 	default <PID extends IResourcePersistentId<?>> Stream<PID> searchForIdStream(
 			SearchParameterMap theParams,
@@ -384,6 +384,11 @@ public interface IFhirResourceDao<T extends IBaseResource> extends IDao {
 		return iResourcePersistentIds.stream();
 	}
 
+	default <PID extends IResourcePersistentId<?>> Stream<PID> searchForIdStream(
+			SearchParameterMap theParams, RequestDetails theRequest) {
+		return searchForIdStream(theParams, theRequest, null);
+	}
+
 	/**
 	 * Takes a map of incoming raw search parameters and translates/parses them into
 	 * appropriate {@link IQueryParameterType} instances of the appropriate type

hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/dao/tx/HapiTransactionService.java

@@ -104,12 +104,16 @@ public class HapiTransactionService implements IHapiTransactionService {
 
 	@Override
 	public IExecutionBuilder withRequest(@Nullable RequestDetails theRequestDetails) {
-		return new ExecutionBuilder(theRequestDetails);
+		return buildExecutionBuilder(theRequestDetails);
 	}
 
 	@Override
 	public IExecutionBuilder withSystemRequest() {
-		return new ExecutionBuilder(null);
+		return buildExecutionBuilder(null);
+	}
+
+	protected IExecutionBuilder buildExecutionBuilder(@Nullable RequestDetails theRequestDetails) {
+		return new ExecutionBuilder(theRequestDetails);
 	}
 
 	/**
@@ -236,15 +240,7 @@ public class HapiTransactionService implements IHapiTransactionService {
 
 	@Nullable
 	protected <T> T doExecute(ExecutionBuilder theExecutionBuilder, TransactionCallback<T> theCallback) {
-		final RequestPartitionId requestPartitionId;
+		final RequestPartitionId requestPartitionId = theExecutionBuilder.getEffectiveRequestPartitionId();
-		if (theExecutionBuilder.myRequestPartitionId != null) {
-			requestPartitionId = theExecutionBuilder.myRequestPartitionId;
-		} else if (theExecutionBuilder.myRequestDetails != null) {
-			requestPartitionId = myRequestPartitionHelperSvc.determineGenericPartitionForRequest(
-					theExecutionBuilder.myRequestDetails);
-		} else {
-			requestPartitionId = null;
-		}
 		RequestPartitionId previousRequestPartitionId = null;
 		if (requestPartitionId != null) {
 			previousRequestPartitionId = ourRequestPartitionThreadLocal.get();
@@ -443,7 +439,8 @@ public class HapiTransactionService implements IHapiTransactionService {
 		}
 	}
 
-	protected class ExecutionBuilder implements IExecutionBuilder, TransactionOperations {
+	// wipmb is Clone ok, or do we want an explicit copy constructor?
+	protected class ExecutionBuilder implements IExecutionBuilder, TransactionOperations, Cloneable {
 
 		private final RequestDetails myRequestDetails;
 		private Isolation myIsolation;
@@ -451,7 +448,7 @@ public class HapiTransactionService implements IHapiTransactionService {
 		private boolean myReadOnly;
 		private TransactionDetails myTransactionDetails;
 		private Runnable myOnRollback;
-		private RequestPartitionId myRequestPartitionId;
+		protected RequestPartitionId myRequestPartitionId;
 
 		protected ExecutionBuilder(RequestDetails theRequestDetails) {
 			myRequestDetails = theRequestDetails;
@@ -533,6 +530,19 @@ public class HapiTransactionService implements IHapiTransactionService {
 		public Propagation getPropagation() {
 			return myPropagation;
 		}
+
+		@Nullable
+		protected RequestPartitionId getEffectiveRequestPartitionId() {
+			final RequestPartitionId requestPartitionId;
+			if (myRequestPartitionId != null) {
+				requestPartitionId = myRequestPartitionId;
+			} else if (myRequestDetails != null) {
+				requestPartitionId = myRequestPartitionHelperSvc.determineGenericPartitionForRequest(myRequestDetails);
+			} else {
+				requestPartitionId = null;
+			}
+			return requestPartitionId;
+		}
 	}
 
 	/**

hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/dao/tx/IHapiTransactionService.java

@@ -31,6 +31,7 @@ import org.springframework.transaction.support.TransactionCallback;
 import org.springframework.transaction.support.TransactionOperations;
 
 import java.util.concurrent.Callable;
+import java.util.stream.Stream;
 
 /**
  * This class is used to execute code within the context of a database transaction,
@@ -107,5 +108,13 @@ public interface IHapiTransactionService {
 		<T> T execute(Callable<T> theTask);
 
 		<T> T execute(@Nonnull TransactionCallback<T> callback);
+
+		default <T> T read(Callable<T> theCallback) {
+			return execute(theCallback);
+		}
+
+		default <T> Stream<T> search(Callable<Stream<T>> theCallback) {
+			return execute(theCallback);
+		}
 	}
 }

hapi-fhir-test-utilities/src/main/java/ca/uhn/test/util/LogbackCaptureTestExtension.java

@@ -23,13 +23,13 @@ import ch.qos.logback.classic.Level;
 import ch.qos.logback.classic.Logger;
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
+import jakarta.annotation.Nonnull;
 import org.hamcrest.Matcher;
 import org.junit.jupiter.api.extension.AfterEachCallback;
 import org.junit.jupiter.api.extension.BeforeEachCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
 import org.slf4j.LoggerFactory;
 
-import jakarta.annotation.Nonnull;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Predicate;
@@ -37,7 +37,6 @@ import java.util.stream.Collectors;
 
 /**
  * Test helper to collect logback lines.
- *
  * The empty constructor will capture all log events, or you can name a log root to limit the noise.
  */
 public class LogbackCaptureTestExtension implements BeforeEachCallback, AfterEachCallback {
@@ -83,6 +82,10 @@ public class LogbackCaptureTestExtension implements BeforeEachCallback, AfterEac
 		this((Logger) LoggerFactory.getLogger(theLoggerName), theLevel);
 	}
 
+    public LogbackCaptureTestExtension(Class<?> theClass) {
+		this(theClass.getName());
+    }
+
     /**
 	 * Returns a copy to avoid concurrent modification errors.
 	 * @return A copy of the log events so far.

snapshot-pipeline.yml

@@ -47,8 +47,8 @@ steps:
 
   # 3. Import keys into gpg
   - bash: |
-      apt update
+      sudo apt update
-      apt install -y gpg
+      sudo apt install -y gpg
       gpg --import --no-tty --batch --yes $(Agent.TempDirectory)/public.key
       gpg --import --no-tty --batch --yes $(Agent.TempDirectory)/private.key
       gpg --list-keys --keyid-format LONG