com.mulesoft.mule.runtime.module.batch.internal.engine

Class DefaultBatchEngine

java.lang.Object

com.mulesoft.mule.runtime.module.batch.internal.BaseBatchProcessingListenerOwner

com.mulesoft.mule.runtime.module.batch.internal.engine.DefaultBatchEngine

All Implemented Interfaces:

BatchProcessingListenersOwner, BatchEngine, org.mule.runtime.api.lifecycle.Initialisable, org.mule.runtime.api.lifecycle.Startable, org.mule.runtime.api.lifecycle.Stoppable, org.mule.runtime.core.api.context.MuleContextAware

public class DefaultBatchEngine
extends BaseBatchProcessingListenerOwner
implements BatchEngine, org.mule.runtime.core.api.context.MuleContextAware, org.mule.runtime.api.lifecycle.Initialisable, org.mule.runtime.api.lifecycle.Stoppable, org.mule.runtime.api.lifecycle.Startable

Default implementation of BatchEngine

Since:

3.5.0

Field Summary

Fields inherited from interface org.mule.runtime.api.lifecycle.Initialisable

PHASE_NAME

Fields inherited from interface org.mule.runtime.api.lifecycle.Stoppable

PHASE_NAME

Fields inherited from interface org.mule.runtime.api.lifecycle.Startable

PHASE_NAME

Constructor Summary

Constructors

Constructor and Description
DefaultBatchEngine()

Method Summary

Modifier and Type	Method and Description
static String	buildExceptionLogMessage(Throwable t) Attempts top get meaningful information from an exception and build a message from it.
void	cancel(BatchJobInstanceAdapter jobInstance) Stops the jobInstance if it's executing (by delegating into BatchEngine.stopExecution(BatchJobInstanceAdapter, boolean).
void	cancelAllRunningInstance() Invokes BatchEngine.cancel(BatchJobInstanceAdapter) on every running instance of any job
org.mule.runtime.core.api.event.CoreEvent.Builder	createEventBuilder(Record record, BatchJobInstanceAdapter jobInstance) Creates a new CoreEvent.Builder with the base attributes set for a CoreEvent to process the given record for the given jobInstance.
org.mule.runtime.core.api.event.CoreEvent.Builder	createEventBuilder(Record record, BatchJobInstanceAdapter jobInstance, CompletableFuture<Void> externalCompletion) Creates a new CoreEvent.Builder with the base attributes set for a CoreEvent to process the given record for the given jobInstance.
BatchJobInstanceAdapter	createNewJobInstance(BatchJobAdapter job, org.mule.runtime.core.api.event.CoreEvent event) Creates and initialises a new BatchJobInstanceAdapter related to the given job.
BatchTransactionContext	createTransactionContext(BatchJobInstanceAdapter jobInstance) Creates a new BatchTransactionContext referencing the given {@link BatchJobInstanceAdapter}
BatchJobInstanceAdapter	finishExecution(BatchJobInstanceAdapter jobInstance, boolean shouldRefresh) Finishes execution of the BatchJobInstanceAdapter in a successful way and makes it no longer eligible for execution by the steps.
Collection<BatchJob>	getBatchJobs() Returns all defined BatchJobs
BatchQueueManager	getBatchQueueManager() Returns the BatchQueueManager in which queueing operations are delegated
BatchTransactionContextFactory	getBatchTransactionContextFactory() For monitoring and testing purposes only
List<Record>	getBlockFrom(BatchTransactionContext context) Returns the records in the stepping queue which corresponds to the BatchJobInstance referenced by the given context
int	getBlockSize(com.mulesoft.mule.runtime.module.batch.api.BatchJobInstance jobInstance) Returns the BatchJobAdapter.getBlockSize() for the BatchJob that owns the given jobInstance
Optional<BatchJob>	getJob(String jobName)
BatchJobAdapter	getJobFor(com.mulesoft.mule.runtime.module.batch.api.BatchJobInstance jobInstance) Returns the owning BatchJobAdapter for the given BatchJobInstance
BatchJobInstanceStore	getJobInstanceStore() Returns the BatchJobInstanceStore in which persistence operations are delegated on
Lock	getLock(BatchJobInstanceAdapter jobInstance) Obtains a Lock for the given jobInstance.
long	getSteppingQueueSize(BatchTransactionContext context) Returns the amount of records in the stepping queue which corresponds to the BatchJobInstance referenced by the given context
void	initialise()
BatchJobInstanceAdapter	load(BatchJobInstanceAdapter jobInstance, org.mule.runtime.core.api.event.CoreEvent event) Splits the CoreEvent payload and loads it into the BatchJobInstanceAdapter queue.
void	registerBatchJob(BatchJobAdapter batchJob)
void	releaseResources(BatchJobInstanceAdapter jobInstance, boolean includeFailedRecords) Releases the resources still allocated to the given BatchJobInstanceAdapter, including queues and storage
void	resumeExecution(BatchJobInstanceAdapter jobInstance) Resumes execution of a BatchJobInstance which was stopped through the BatchEngine.stopExecution(BatchJobInstanceAdapter, boolean) method.
void	setBatchQueueManager(BatchQueueManager batchQueueManager)
void	setHistoryExpirationFrequency(long historyExpirationFrequency)
void	setJobInstanceStore(BatchJobInstanceStore jobInstanceFactory)
void	setMuleContext(org.mule.runtime.core.api.MuleContext context)
void	start()
void	startExecution(BatchJobInstanceAdapter jobInstance) Starts asynchronous execution of the given BatchJobInstanceAdapter.
void	stop()
void	stopExecution(BatchJobInstanceAdapter jobInstanceAdapter, boolean markAsFailed) Stops the execution of the given BatchJobInstanceAdapter.
BatchJobInstanceAdapter	updateStatisticsAndRoute(BatchTransactionContext ctx, List<Record> records)

Methods inherited from class com.mulesoft.mule.runtime.module.batch.internal.BaseBatchProcessingListenerOwner

addBatchProcessingListener, addBatchProcessingListener, getListeners, removeBatchProcessingListener, removeBatchProcessingListener

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface com.mulesoft.mule.runtime.module.batch.BatchProcessingListenersOwner

addBatchProcessingListener, addBatchProcessingListener, removeBatchProcessingListener, removeBatchProcessingListener

Constructor Detail

DefaultBatchEngine

public DefaultBatchEngine()

Method Detail

initialise

public void initialise()
                throws org.mule.runtime.api.lifecycle.InitialisationException

Specified by:

initialise in interface org.mule.runtime.api.lifecycle.Initialisable

Throws:

org.mule.runtime.api.lifecycle.InitialisationException

stop

public void stop()
          throws org.mule.runtime.api.exception.MuleException

Specified by:

stop in interface org.mule.runtime.api.lifecycle.Stoppable

Throws:

org.mule.runtime.api.exception.MuleException

createNewJobInstance

public BatchJobInstanceAdapter createNewJobInstance(BatchJobAdapter job,
                                                    org.mule.runtime.core.api.event.CoreEvent event)
                                             throws org.mule.runtime.api.exception.MuleException

Creates and initialises a new BatchJobInstanceAdapter related to the given job. The returned instance will be already persisted

Specified by:

createNewJobInstance in interface BatchEngine

Parameters:

job - the BatchJobAdapter being executed

event - the CoreEvent that triggered the job's execution

Returns:

a new persistent instance of BatchJobInstanceAdapter

Throws:

org.mule.runtime.api.exception.MuleException

createTransactionContext

public BatchTransactionContext createTransactionContext(BatchJobInstanceAdapter jobInstance)

Creates a new BatchTransactionContext referencing the given {@link BatchJobInstanceAdapter}

Specified by:

createTransactionContext in interface BatchEngine

Parameters:

jobInstance - a not null BatchJobInstanceAdapter

Returns:

a new BatchTransactionContext

load

public BatchJobInstanceAdapter load(BatchJobInstanceAdapter jobInstance,
                                    org.mule.runtime.core.api.event.CoreEvent event)
                             throws org.mule.runtime.api.exception.MuleException

Splits the CoreEvent payload and loads it into the BatchJobInstanceAdapter queue. If this process fails, the job instance's resources are released.

Specified by:

load in interface BatchEngine

Parameters:

jobInstance - a BatchJobInstanceAdapter

event - a CoreEvent which payload is splittable

Returns:

a BatchJobInstanceAdapter which state has been updated to reflect this operation's result

Throws:

org.mule.runtime.api.exception.MuleException

startExecution

public void startExecution(BatchJobInstanceAdapter jobInstance)
                    throws org.mule.runtime.api.exception.MuleException

Starts asynchronous execution of the given BatchJobInstanceAdapter. The instance will be persisted at this point and will become elegible for execution by the steps. It also fires a BatchNotification.JOB_PROCESS_RECORDS_BEGIN notification

Specified by:

startExecution in interface BatchEngine

Parameters:

jobInstance - the BatchJobInstanceAdapter to be executed

Throws:

org.mule.runtime.api.exception.MuleException

finishExecution

public BatchJobInstanceAdapter finishExecution(BatchJobInstanceAdapter jobInstance,
                                               boolean shouldRefresh)
                                        throws org.mule.runtime.api.exception.MuleException

Finishes execution of the BatchJobInstanceAdapter in a successful way and makes it no longer eligible for execution by the steps. If the owning BatchJob has an onComplete phase then it's executed. Resources associated to the jobInstance are released by invoking BatchEngine.releaseResources(BatchJobInstanceAdapter, boolean). Finally, it fires a BatchNotification.JOB_SUCCESSFUL

Specified by:

finishExecution in interface BatchEngine

Parameters:

jobInstance - an instance ofBatchJobInstanceAdapter}

shouldRefresh - if true then the jobInstance is refreshed before actually performing the logic

Returns:

the given jobInstance, refreshed if shouldRefresh was true

Throws:

org.mule.runtime.api.exception.MuleException

releaseResources

public void releaseResources(BatchJobInstanceAdapter jobInstance,
                             boolean includeFailedRecords)

Releases the resources still allocated to the given BatchJobInstanceAdapter, including queues and storage

Specified by:

releaseResources in interface BatchEngine

Parameters:

jobInstance - a BatchJobInstanceAdapter

includeFailedRecords - if true, the stored failed records will also be wiped out.

createEventBuilder

public org.mule.runtime.core.api.event.CoreEvent.Builder createEventBuilder(Record record,
                                                                            BatchJobInstanceAdapter jobInstance)

Creates a new CoreEvent.Builder with the base attributes set for a CoreEvent to process the given record for the given jobInstance. This method will create a copy of the original event that generated the jobInstance, will set the record's payload as the message payload and will put the record in an invocation property under the key in BatchProperties.RECORD_VARIABLE_NAME

Specified by:

createEventBuilder in interface BatchEngine

Parameters:

record - an instance of Record

jobInstance - an instance of BatchJobInstanceAdapter

Returns:

a new instance of CoreEvent.Builder

createEventBuilder

public org.mule.runtime.core.api.event.CoreEvent.Builder createEventBuilder(Record record,
                                                                            BatchJobInstanceAdapter jobInstance,
                                                                            CompletableFuture<Void> externalCompletion)

Creates a new CoreEvent.Builder with the base attributes set for a CoreEvent to process the given record for the given jobInstance. This method will create a copy of the original event that generated the jobInstance, will set the record's payload as the message payload and will put the record in an invocation property under the key in BatchProperties.RECORD_VARIABLE_NAME

Specified by:

createEventBuilder in interface BatchEngine

Parameters:

record - an instance of Record

jobInstance - an instance of BatchJobInstanceAdapter

externalCompletion - A custom completion callback for the event context. If null, then default completion will be used

Returns:

a new instance of CoreEvent.Builder

stopExecution

public void stopExecution(BatchJobInstanceAdapter jobInstanceAdapter,
                          boolean markAsFailed)
                   throws org.mule.runtime.api.exception.MuleException

Stops the execution of the given BatchJobInstanceAdapter. This doesn't mean that all threads processing records will immediately stop. It means that no more records will be taken out of the queue and start processing. All records currently being processed will complete their normal execution. if markAsFailed is true the jobInstance status will be set to BatchJobInstanceStatus.FAILED_PROCESS_RECORDS and a BatchNotification.JOB_PROCESS_RECORDS_FAILED notification will be fired. Otherwise, the status will be set to BatchJobInstanceStatus.STOPPED and a BatchNotification.JOB_STOPPED notification is fired instead.

Specified by:

stopExecution in interface BatchEngine

Parameters:

jobInstanceAdapter - a BatchJobInstanceAdapter

markAsFailed - if true the jobInstance status will be set to BatchJobInstanceStatus.FAILED_PROCESS_RECORDS. Otherwise it will be set to BatchNotification.JOB_STOPPED

Throws:

org.mule.runtime.api.exception.MuleException

resumeExecution

public void resumeExecution(BatchJobInstanceAdapter jobInstance)
                     throws org.mule.runtime.api.exception.MuleException

Resumes execution of a BatchJobInstance which was stopped through the BatchEngine.stopExecution(BatchJobInstanceAdapter, boolean) method.

Execution is resumed by setting the job instance status back to BatchJobInstanceStatus.EXECUTING. The BatchNotification.JOB_PROCESS_RECORDS_BEGIN notification will be fired again.

Specified by:

resumeExecution in interface BatchEngine

Parameters:

jobInstance - the BatchJobInstanceAdapter to be resmed

Throws:

org.mule.runtime.api.exception.MuleException - in case of unhandled error

cancel

public void cancel(BatchJobInstanceAdapter jobInstance)
            throws org.mule.runtime.api.exception.MuleException

Stops the jobInstance if it's executing (by delegating into BatchEngine.stopExecution(BatchJobInstanceAdapter, boolean). Finally, the instance's resources are released (by delegating into BatchEngine.releaseResources(BatchJobInstanceAdapter, boolean) and a BatchNotification.JOB_CANCELLED notification is fired

Specified by:

cancel in interface BatchEngine

Parameters:

jobInstance - a BatchJobInstanceAdapter

Throws:

org.mule.runtime.api.exception.MuleException

cancelAllRunningInstance

public void cancelAllRunningInstance()
                              throws org.mule.runtime.api.exception.MuleException

Invokes BatchEngine.cancel(BatchJobInstanceAdapter) on every running instance of any job

Specified by:

cancelAllRunningInstance in interface BatchEngine

Throws:

org.mule.runtime.api.exception.MuleException

getSteppingQueueSize

public long getSteppingQueueSize(BatchTransactionContext context)

Description copied from interface: BatchEngine

Returns the amount of records in the stepping queue which corresponds to the BatchJobInstance referenced by the given context

Specified by:

getSteppingQueueSize in interface BatchEngine

Parameters:

context - the context for a current transaction

Returns:

the number of queued records. Will be greater or equal to zero

getBlockFrom

public List<Record> getBlockFrom(BatchTransactionContext context)

Description copied from interface: BatchEngine

Returns the records in the stepping queue which corresponds to the BatchJobInstance referenced by the given context

Specified by:

getBlockFrom in interface BatchEngine

Parameters:

context - the context for a current transaction

Returns:

the queued records.

updateStatisticsAndRoute

public BatchJobInstanceAdapter updateStatisticsAndRoute(BatchTransactionContext ctx,
                                                        List<Record> records)
                                                 throws org.mule.runtime.api.exception.MuleException

Specified by:

updateStatisticsAndRoute in interface BatchEngine

Throws:

org.mule.runtime.api.exception.MuleException

getLock

public Lock getLock(BatchJobInstanceAdapter jobInstance)

Obtains a Lock for the given jobInstance. Invoking this method with difference instances which share the same id, will always return the same lock

Specified by:

getLock in interface BatchEngine

Parameters:

jobInstance - a BatchJobInstanceAdapter

Returns:

a Lock

getJobFor

public BatchJobAdapter getJobFor(com.mulesoft.mule.runtime.module.batch.api.BatchJobInstance jobInstance)

Returns the owning BatchJobAdapter for the given BatchJobInstance

Specified by:

getJobFor in interface BatchEngine

Parameters:

jobInstance - a BatchJobInstance

Returns:

a BatchJobAdapter

registerBatchJob

public void registerBatchJob(BatchJobAdapter batchJob)

Specified by:

registerBatchJob in interface BatchEngine

getBatchJobs

public Collection<BatchJob> getBatchJobs()

Returns all defined BatchJobs

Specified by:

getBatchJobs in interface BatchEngine

Returns:

a Collection of BatchJob. The collection might be empty but will never be null

getJob

public Optional<BatchJob> getJob(String jobName)

Specified by:

getJob in interface BatchEngine

getJobInstanceStore

public BatchJobInstanceStore getJobInstanceStore()

Returns the BatchJobInstanceStore in which persistence operations are delegated on

Specified by:

getJobInstanceStore in interface BatchEngine

Returns:

a BatchJobInstance

setJobInstanceStore

public void setJobInstanceStore(BatchJobInstanceStore jobInstanceFactory)

getBatchQueueManager

public BatchQueueManager getBatchQueueManager()

Returns the BatchQueueManager in which queueing operations are delegated

Specified by:

getBatchQueueManager in interface BatchEngine

Returns:

a BatchQueueManager

setBatchQueueManager

public void setBatchQueueManager(BatchQueueManager batchQueueManager)

setMuleContext

public void setMuleContext(org.mule.runtime.core.api.MuleContext context)

Specified by:

setMuleContext in interface org.mule.runtime.core.api.context.MuleContextAware

getBlockSize

public int getBlockSize(com.mulesoft.mule.runtime.module.batch.api.BatchJobInstance jobInstance)

Returns the BatchJobAdapter.getBlockSize() for the BatchJob that owns the given jobInstance

Specified by:

getBlockSize in interface BatchEngine

Parameters:

jobInstance - a BatchJobInstance

Returns:

the owner job's block size

getBatchTransactionContextFactory

public BatchTransactionContextFactory getBatchTransactionContextFactory()

For monitoring and testing purposes only

Specified by:

getBatchTransactionContextFactory in interface BatchEngine

Returns:

the current BatchTransactionContextFactory

setHistoryExpirationFrequency

public void setHistoryExpirationFrequency(long historyExpirationFrequency)

buildExceptionLogMessage

public static String buildExceptionLogMessage(Throwable t)

Attempts top get meaningful information from an exception and build a message from it.

Parameters:

t - the exception to build a log-able message for.

Returns:

the built message to log.

start

public void start()
           throws org.mule.runtime.api.exception.MuleException

Specified by:

start in interface org.mule.runtime.api.lifecycle.Startable

Throws:

org.mule.runtime.api.exception.MuleException