SMILA/Documentation/Worker/PipelineProcessorWorker

Available since SMILA 0.9.0!

PipelineProcessingWorker (bundle org.eclipse.smila.processing.worker)

The PipelineProcessingWorker is a worker designed to process synchronous pipelines inside a asynchronous workflow. The worker in principal is independant of a dedicated pipeline processing implementation. However, in SMILA we use BPEL pipelines for synchronous workflows, so in common speech the worker is also called BPEL worker.

BPEL pipelines resp. pipelets are able to process records in parallel. Therefore the PipelineProcessingWorker can divide the records of the input bulk in bunches of records to be processed in parallel. This can be configured via numberOfParallelRecords parameter (see below).

JavaDoc

This page gives only a rough overview of the service. Please refer to the JavaDoc for detailed information about the Java components.

Configuration

The PipelineProcessingWorker is configured via incoming task parameters. These parameters could have been set e.g. in a job definition.

Parameter	Description	Default value
pipelineName	name of the synchronous (BPEL) pipeline to execute	---
numberOfParallelRecords	number of records to be processed in parallel by the synchronous workflow (value <= 0 -> use default value)	1

Sample job definition that sets the parameters:

{
  "name":"myJob",
  "parameters":{
    "pipelineName": "myBpelPipeline",
    "numberOfParallelRecords": "10",
    ...
   },
  "workflow":"myWorkflow"
}

PipelineProcessingWorker definition in workers.json

  { "name": "pipelineProcessingWorker",
     "parameters": [ "pipelineName" ],
     "input": [ 
         {  "name": "input",              
            "type": "recordBulks"
         } ],
     "output": [ 
         {  "name": "output",
            "type": "recordBulks",
            "modes": ["optional"]
         } ]             
  }

Error handling

The following errors may occur when a task for the PipelineProcessingWorker

• No (resp. invalid) pipeline parameter
  • if the given pipeline parameter is not set (or invalid) the task will fail with a non-recoverable error
• ProcessingException while processing bunch of parallel records.
  • recoverable ProcessingException: the current task will fail with a recoverable error, so the whole task (with all records) will be repeated.
  • non-recoverable ProcessingException: an error will be logged and the worker will continue with the next bunch of records. The records of the current bunch will be lost. (This is implemented in that way to not fail the whole task (with the whole input record bulk) if a single record is defect.)