Package ca.uhn.fhir.batch2.api

Interface IJobPersistence

All Superinterfaces:

IWorkChunkPersistence

public interface IJobPersistence
extends IWorkChunkPersistence

Some of this is tested in ca.uhn.hapi.fhir.batch2.test.AbstractIJobPersistenceSpecificationTest This is a transactional interface, but we have pushed the declaration of calls that have @Transactional(propagation = Propagation.REQUIRES_NEW) down to the implementations since we have a synchronized wrapper that was double-creating the NEW transaction.

Nested Class Summary

Nested Classes

Modifier and Type	Interface	Description
static class	IJobPersistence.CreateResult
static interface	IJobPersistence.JobInstanceUpdateCallback	Callback to update a JobInstance within a locked transaction.

Field Summary

Fields

Modifier and Type	Field	Description
static final org.slf4j.Logger	ourLog

Method Summary

Modifier and Type	Method	Description
boolean	advanceJobStepAndUpdateChunkStatus(String theJobInstanceId, String theNextStepId, boolean theIsReductionStepBoolean)	Atomically advance the given job to the given step and change the status of all QUEUED and GATE_WAITING chunks in the next step to READY
JobOperationResultJson	cancelInstance(String theInstanceId)	Marks an instance as cancelled
WorkChunk	createWorkChunk(WorkChunk theWorkChunk)
void	deleteChunksAndMarkInstanceAsChunksPurged(String theInstanceId)	Deletes all work chunks associated with the instance
void	deleteInstanceAndChunks(String theInstanceId)	Deletes the instance and all associated work chunks
void	enqueueWorkChunkForProcessing(String theChunkId, Consumer<Integer> theCallback)
org.springframework.data.domain.Page<WorkChunkMetadata>	fetchAllWorkChunkMetadataForJobInStates(org.springframework.data.domain.Pageable thePageable, String theInstanceId, Set<WorkChunkStatusEnum> theStates)	Fetches an iterator that retrieves WorkChunkMetadata from the db.
Stream<WorkChunk>	fetchAllWorkChunksForStepStream(String theInstanceId, String theStepId)	Fetch all chunks with data for a given instance for a given step id - read-only.
Iterator<WorkChunk>	fetchAllWorkChunksIterator(String theInstanceId, boolean theWithData)	Fetch all chunks for a given instance.
Optional<JobInstance>	fetchInstance(String theInstanceId)	Fetch an instance
List<JobInstance>	fetchInstances(int thePageSize, int thePageIndex)	Fetch all instances
List<JobInstance>	fetchInstances(FetchJobInstancesRequest theRequest, int theStart, int theBatchSize)	Fetches any existing jobs matching provided request parameters
List<JobInstance>	fetchInstances(String theJobDefinitionId, Set<StatusEnum> theStatuses, Date theCutoff, org.springframework.data.domain.Pageable thePageable)
List<JobInstance>	fetchInstancesByJobDefinitionId(String theJobDefinitionId, int theCount, int theStart)	Fetch all job instances for a given job definition id
List<JobInstance>	fetchInstancesByJobDefinitionIdAndStatus(String theJobDefinitionId, Set<StatusEnum> theRequestedStatuses, int thePageSize, int thePageIndex)
org.springframework.data.domain.Page<JobInstance>	fetchJobInstances(JobInstanceFetchRequest theRequest)	Fetches all job instances based on the JobFetchRequest
List<JobInstance>	fetchRecentInstances(int thePageSize, int thePageIndex)	Fetch instances ordered by myCreateTime DESC
Set<WorkChunkStatusEnum>	getDistinctWorkChunkStatesForJobAndStep(String theInstanceId, String theCurrentStepId)	Returns set of all distinct states for the specified job instance id and step id.
boolean	markInstanceAsStatusWhenStatusIn(String theInstance, StatusEnum theStatusEnum, Set<StatusEnum> thePriorStates)
default boolean	onChunkDequeued(String theJobInstanceId)	Move from QUEUED->IN_PROGRESS when a work chunk arrives.
default IJobPersistence.CreateResult	onCreateWithFirstChunk(JobDefinition<?> theJobDefinition, String theParameters)	Create the job, and it's first chunk.
String	storeNewInstance(JobInstance theInstance)	Store a new job instance.
boolean	updateInstance(String theInstanceId, IJobPersistence.JobInstanceUpdateCallback theModifier)	Brute-force hack for now to create a tx boundary - takes a write-lock on the instance while the theModifier runs.
void	updateInstanceUpdateTime(String theInstanceId)
int	updatePollWaitingChunksForJobIfReady(String theInstanceId)	Updates all Work Chunks in POLL_WAITING if their nextPollTime <= now for the given Job Instance.

Methods inherited from interface ca.uhn.fhir.batch2.api.IWorkChunkPersistence

markWorkChunksWithStatusAndWipeData, onWorkChunkCompletion, onWorkChunkCreate, onWorkChunkDequeue, onWorkChunkError, onWorkChunkFailed, onWorkChunkPollDelay

Field Details

ourLog

static final org.slf4j.Logger ourLog

Method Details

storeNewInstance

@Transactional(propagation=REQUIRED)
String storeNewInstance(JobInstance theInstance)

Store a new job instance. This will be called when a new job instance is being kicked off.

Parameters:

theInstance - The details

fetchInstance

Optional<JobInstance> fetchInstance(String theInstanceId)

Fetch an instance

Parameters:

theInstanceId - The instance ID

fetchInstances

List<JobInstance> fetchInstances(String theJobDefinitionId,
 Set<StatusEnum> theStatuses,
 Date theCutoff,
 org.springframework.data.domain.Pageable thePageable)

fetchInstances

List<JobInstance> fetchInstances(FetchJobInstancesRequest theRequest,
 int theStart,
 int theBatchSize)

Fetches any existing jobs matching provided request parameters

fetchInstances

List<JobInstance> fetchInstances(int thePageSize,
 int thePageIndex)

Fetch all instances

enqueueWorkChunkForProcessing

@Transactional(propagation=REQUIRES_NEW)
void enqueueWorkChunkForProcessing(String theChunkId,
 Consumer<Integer> theCallback)

updatePollWaitingChunksForJobIfReady

@Transactional
int updatePollWaitingChunksForJobIfReady(String theInstanceId)

Updates all Work Chunks in POLL_WAITING if their nextPollTime <= now for the given Job Instance.

Parameters:

theInstanceId - the instance id

Returns:

the number of updated chunks

fetchRecentInstances

List<JobInstance> fetchRecentInstances(int thePageSize,
 int thePageIndex)

Fetch instances ordered by myCreateTime DESC

fetchInstancesByJobDefinitionIdAndStatus

List<JobInstance> fetchInstancesByJobDefinitionIdAndStatus(String theJobDefinitionId,
 Set<StatusEnum> theRequestedStatuses,
 int thePageSize,
 int thePageIndex)

fetchInstancesByJobDefinitionId

List<JobInstance> fetchInstancesByJobDefinitionId(String theJobDefinitionId,
 int theCount,
 int theStart)

Fetch all job instances for a given job definition id

fetchJobInstances

org.springframework.data.domain.Page<JobInstance> fetchJobInstances(JobInstanceFetchRequest theRequest)

Fetches all job instances based on the JobFetchRequest

Parameters:

theRequest - - the job fetch request

Returns:

- a page of job instances

getDistinctWorkChunkStatesForJobAndStep

@Transactional
Set<WorkChunkStatusEnum> getDistinctWorkChunkStatesForJobAndStep(String theInstanceId,
 String theCurrentStepId)

Returns set of all distinct states for the specified job instance id and step id.

fetchAllWorkChunksIterator

Iterator<WorkChunk> fetchAllWorkChunksIterator(String theInstanceId,
 boolean theWithData)

Fetch all chunks for a given instance.

Specified by:

fetchAllWorkChunksIterator in interface IWorkChunkPersistence

Parameters:

theInstanceId - - instance id

theWithData - - whether or not to include the data

Returns:

- an iterator for fetching work chunks

fetchAllWorkChunksForStepStream

Stream<WorkChunk> fetchAllWorkChunksForStepStream(String theInstanceId,
 String theStepId)

Fetch all chunks with data for a given instance for a given step id - read-only.

Specified by:

fetchAllWorkChunksForStepStream in interface IWorkChunkPersistence

Returns:

- a stream for fetching work chunks

fetchAllWorkChunkMetadataForJobInStates

@Transactional(propagation=SUPPORTS)
org.springframework.data.domain.Page<WorkChunkMetadata> fetchAllWorkChunkMetadataForJobInStates(org.springframework.data.domain.Pageable thePageable,
 String theInstanceId,
 Set<WorkChunkStatusEnum> theStates)

Fetches an iterator that retrieves WorkChunkMetadata from the db.

Parameters:

theInstanceId - instance id of job of interest

theStates - states of interset

Returns:

an iterator for the workchunks

updateInstance

@Transactional
boolean updateInstance(String theInstanceId,
 IJobPersistence.JobInstanceUpdateCallback theModifier)

Brute-force hack for now to create a tx boundary - takes a write-lock on the instance while the theModifier runs. Keep the callback short to keep the lock-time short. If the status is changing, use JobInstanceStatusUpdater inside theModifier to ensure state-change callbacks are invoked properly.

Parameters:

theInstanceId - the id of the instance to modify

theModifier - a hook to modify the instance - return true to finish the record write

Returns:

true if the instance was modified

deleteInstanceAndChunks

void deleteInstanceAndChunks(String theInstanceId)

Deletes the instance and all associated work chunks

Parameters:

theInstanceId - The instance ID

deleteChunksAndMarkInstanceAsChunksPurged

void deleteChunksAndMarkInstanceAsChunksPurged(String theInstanceId)

Deletes all work chunks associated with the instance

Parameters:

theInstanceId - The instance ID

markInstanceAsStatusWhenStatusIn

@Transactional(propagation=MANDATORY)
boolean markInstanceAsStatusWhenStatusIn(String theInstance,
 StatusEnum theStatusEnum,
 Set<StatusEnum> thePriorStates)

cancelInstance

JobOperationResultJson cancelInstance(String theInstanceId)

Marks an instance as cancelled

Parameters:

theInstanceId - The instance ID

updateInstanceUpdateTime

@Transactional(propagation=MANDATORY)
void updateInstanceUpdateTime(String theInstanceId)

onCreateWithFirstChunk

@Nonnull
@Transactional(propagation=MANDATORY)
default IJobPersistence.CreateResult onCreateWithFirstChunk(JobDefinition<?> theJobDefinition,
 String theParameters)

Create the job, and it's first chunk. We create the chunk atomically with the job so that we never have a state with zero unfinished chunks left until the job is complete. Makes the maintenance run simpler.

Parameters:

theJobDefinition - what kind of job

theParameters - params for the job

Returns:

the ids of the instance and first chunk

onChunkDequeued

@Transactional(propagation=MANDATORY)
default boolean onChunkDequeued(String theJobInstanceId)

Move from QUEUED->IN_PROGRESS when a work chunk arrives. Ignore other prior states.

Returns:

did the transition happen

createWorkChunk

WorkChunk createWorkChunk(WorkChunk theWorkChunk)

advanceJobStepAndUpdateChunkStatus

@Transactional(propagation=REQUIRES_NEW)
boolean advanceJobStepAndUpdateChunkStatus(String theJobInstanceId,
 String theNextStepId,
 boolean theIsReductionStepBoolean)

Atomically advance the given job to the given step and change the status of all QUEUED and GATE_WAITING chunks in the next step to READY

Parameters:

theJobInstanceId - the id of the job instance to be updated

theNextStepId - the id of the next job step

Returns:

whether any changes were made