honeymoose
/
hapi-fhir
mirror of https://github.com/hapifhir/hapi-fhir.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

last-second-fetch

This commit is contained in:
Tadgh 2022-11-14 20:11:24 -08:00
parent 8c896fce53
commit 92b927020a
3 changed files with 17 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/batch2/JpaJobPersistenceImpl.java
View File

@ -297,7 +297,7 @@ public class JpaJobPersistenceImpl implements IJobPersistence {
}); });
} }
private void fetchChunksForStep(String theInstanceId, String theStepId, int thePageSize, int thePageIndex, Consumer<WorkChunk> theConsumer) { public void fetchChunksForStep(String theInstanceId, String theStepId, int thePageSize, int thePageIndex, Consumer<WorkChunk> theConsumer) {
myTxTemplate.executeWithoutResult(tx -> { myTxTemplate.executeWithoutResult(tx -> {
List<Batch2WorkChunkEntity> chunks = myWorkChunkRepository.fetchChunksForStep(PageRequest.of(thePageIndex, thePageSize), theInstanceId, theStepId); List<Batch2WorkChunkEntity> chunks = myWorkChunkRepository.fetchChunksForStep(PageRequest.of(thePageIndex, thePageSize), theInstanceId, theStepId);
for (Batch2WorkChunkEntity chunk : chunks) { for (Batch2WorkChunkEntity chunk : chunks) {

16
hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/api/IJobPersistence.java
View File

@ -35,6 +35,7 @@ import java.util.Iterator;
import java.util.List; import java.util.List;
import java.util.Optional; import java.util.Optional;
import java.util.Set; import java.util.Set;
import java.util.function.Consumer;
public interface IJobPersistence { public interface IJobPersistence {
@ -177,12 +178,15 @@ public interface IJobPersistence {
*/ */
List<WorkChunk> fetchWorkChunksWithoutData(String theInstanceId, int thePageSize, int thePageIndex); List<WorkChunk> fetchWorkChunksWithoutData(String theInstanceId, int thePageSize, int thePageIndex);
/** void fetchChunksForStep(String theInstanceId, String theStepId, int thePageSize, int thePageIndex, Consumer<WorkChunk> theConsumer);
* Fetch all chunks for a given instance.
* @param theInstanceId - instance id
* @param theWithData - whether or not to include the data /**
* @return - an iterator for fetching work chunks * Fetch all chunks for a given instance.
*/ * @param theInstanceId - instance id
* @param theWithData - whether or not to include the data
* @return - an iterator for fetching work chunks
*/
Iterator<WorkChunk> fetchAllWorkChunksIterator(String theInstanceId, boolean theWithData); Iterator<WorkChunk> fetchAllWorkChunksIterator(String theInstanceId, boolean theWithData);
/** /**

8
hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/maintenance/JobInstanceProcessor.java
View File

@ -34,6 +34,7 @@ import ca.uhn.fhir.jpa.batch.log.Logs;
import org.apache.commons.lang3.time.DateUtils; import org.apache.commons.lang3.time.DateUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import java.util.ArrayList;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
@ -175,11 +176,14 @@ public class JobInstanceProcessor {
if (totalChunksForNextStep != queuedChunksForNextStep.size()) { if (totalChunksForNextStep != queuedChunksForNextStep.size()) {
ourLog.error("Total chunk size to submit for next step does not match QUEUED chunk size! [instanceId={}, stepId={}, totalChunks={}, queuedChunks={}]", instanceId, nextStepId, totalChunksForNextStep, queuedChunksForNextStep.size()); ourLog.error("Total chunk size to submit for next step does not match QUEUED chunk size! [instanceId={}, stepId={}, totalChunks={}, queuedChunks={}]", instanceId, nextStepId, totalChunksForNextStep, queuedChunksForNextStep.size());
} }
for (String nextChunkId : queuedChunksForNextStep) { List<String> chunksToSubmit = new ArrayList<>();
myJobPersistence.fetchChunksForStep(instanceId, nextStepId, 10000, 0, chunk -> chunksToSubmit.add(chunk.getId()));
// for (String nextChunkId : queuedChunksForNextStep) {
for (String nextChunkId : chunksToSubmit) {
JobWorkNotification workNotification = new JobWorkNotification(myInstance, nextStepId, nextChunkId); JobWorkNotification workNotification = new JobWorkNotification(myInstance, nextStepId, nextChunkId);
myBatchJobSender.sendWorkChannelMessage(workNotification); myBatchJobSender.sendWorkChannelMessage(workNotification);
} }
ourLog.debug("Submitted a batch of chunks for processing. [chunkCount={}, instanceId={}, stepId={}]", queuedChunksForNextStep.size(), instanceId, nextStepId); ourLog.debug("Submitted a batch of chunks for processing. [chunkCount={}, instanceId={}, stepId={}]", chunksToSubmit.size(), instanceId, nextStepId);
myInstance.setCurrentGatedStepId(nextStepId); myInstance.setCurrentGatedStepId(nextStepId);
myJobPersistence.updateInstance(myInstance); myJobPersistence.updateInstance(myInstance);
} }