1 Overview
2 Key Functionality
- 2.1 Sequence Task Execution
  - 2.1.1 Simple flow execution
  - 2.1.2 Simple flow execution (failed stage)
  - 2.1.3 Simple flow execution (failed stage with cancellation)
- 2.2 Parallel Task Execution
  - 2.2.1 Parallel Stage execution
  - 2.2.2 Parallel Stage execution(stage failed)
  - 2.2.3 Parallel Stage execution (stage failed with cancellation)
  - 2.2.4 Nested flow execution
3 Interfaces and models
- 3.1 Flow / Flow Builder
- 3.2 Flow Engine
  - 3.2.1 Builder
  - 3.2.2 Stage Report Provider
- 3.3 Stage
  - 3.3.1 Flow Methods
  - 3.3.2 Listenable Methods
- 3.4 Parallel Stage
  - 3.4.1 Parallel Stage Builder
- 3.5 Dynamic Stage
- 3.6 Stage Context
- 3.7 Stage executor
4 Logging

Overview

folio-flow-engine is a Java library, that utilizes native classes for Asynchronous execution like CompletableFuture, ExecutorService etc. to provide a friendly interface for composing tasks to be executed with different rules.

GitHub repository: https://github.com/folio-org/folio-flow-engine

Key Functionality

Flow engine provides the following functionality:

Asynchronous execution
Stage execution in sequence
Stage execution in parallel
Complex flow execution, including execution of nested flows in parallel
Dynamic stage generation based on context data
Stage cancellation of one of the upstream stages failed
Stage execution observability
- methods to observe critical moments of stage execution
- explicit logging, including flow engine reports in logs
Java-friendly context management using the Decorator pattern

Sequence Task Execution

Description

The Flow Engine allows tasks to be executed in a specific order, ensuring each task is completed before the next one begins.

Usage

Define a sequence of tasks in the desired order.
The Flow Engine will execute each task one after the other.
If a task fails, subsequent tasks in the sequence will not be executed.

Flow engine follows the following scenarios:

Simple flow execution

A flow engine can execute stages in sequence, waiting for the previous stage to be finished before executing the next one.

Data can be transferred between stages using StageContext model.

Stage context store data with key as String and value as Object. There is a way to make context more Java-friendly using AbstractStageContextWrapper

Flow.builder()
  .id("Simple Flow")
  .stage(task1)
  .stage(task2)
  .stage(task3)
  .build();

Simple flow execution (failed stage)

When one of the stages in flow fails - the flow manager stops the execution and calls onFlowErrorStage (if defined)

Flow.builder()
  .id("Simple Flow")
  .stage(task1)
  .stage(task2)
  .stage(task3)
  .stage(task4)
  .onFlowError(onFlowErrorStage)
  .build();

This behavior is default, and it can be re-configured:

Flow.builder()
  // stages
  .executionStrategy(CANCEL_ON_ERROR | IGNORE_ON_ERROR)
  .build()

Simple flow execution (failed stage with cancellation)

Flow.builder()
  .id("Simple Flow")
  .stage(task1)
  .stage(task2)
  .stage(task3)
  .stage(task4)
  .onFlowCancellation(onFlowCancellationStage)/ta
  .executionStrategy(CANCEL_ON_ERROR)
  .build();

Parallel Task Execution

Description

The Flow Engine supports the execution of multiple tasks simultaneously, leveraging concurrency to improve efficiency and reduce total execution time.

Usage

Define tasks that can be executed in parallel.
The Flow Engine will manage the concurrent execution of these tasks.
- parallelism is configured in the FlowEngine component
If one task fails, all parallel tasks will be canceled except failed ones (if the strategy is CANCEL_ON_ERROR).

Parallel Stage execution

Flow.builder()
  .id("Simple Flow")
  .stage(task1)
  .stage(ParallelStage.of(task2_1, task2_2, task2_3))
  .stage(task3)
  .build();

Note that parallelism is customizable by defining an executor when building FlowEngine component

public FlowEngine flowEngine() {
  return FlowEngine.builder()
    .name("flow-engine")
    .executor(/* custom executor can be defined here */))
    .executionTimeout(/* stage timeout execution can be defined here */)
    .build();
}

Parallel Stage execution(stage failed)

Flow.builder()
  .id("Simple Flow")
  .stage(task1)
  .stage(ParallelStage.of(task2_1, task2_2, task2_3))
  .stage(task3)
  .onFlowError(onFlowErrorStage)
  .build();

Parallel Stage execution (stage failed with cancellation)

Flow.builder()
  .id("Simple Flow")
  .stage(task1)
  .stage(ParallelStage.of(task2_1, task2_2, task2_3))
  .stage(task3)
  .onFlowCancellation(onFlowCancellationStage)
  .executionStrategy(CANCEL_ON_ERROR)
  .build();

Nested flow execution

var flow1 = Flow.builder()
  .id("Flow 1")
  .stage(task2_1_1)
  .stage(task2_1_2)
  .build();

var flow2 = Flow.builder()
  .id("Flow 1")
  .stage(task2_2_1)
  .stage(task2_2_2)
  .build();
  
Flow.builder()
  .id("Simple Flow")
  .stage(task1)
  .stage(ParallelStage.of(flow1, flow2))
  .stage(task3)
  .build();

Interfaces and models

Flow / Flow Builder

Provides a Flow object, containing a sequence of stages and flows to be executed

Sucessfull flow executions are always finished without exceptions

If error is occurred during flow execution and executionStrategy is IGNORE_ON_ERROR - then FlowExecutionException will be raised with a reason in cause
If error is occurred during flow execution and executionStrategy is CANCEL_ON_ERROR - then
- FlowCancelledException will be raised with a reason of cancellation in cause
- FlowCancellationException will be raised with a reason of cancellation failure in cause

Flows can be executed in async mode using flowEngine.executeAsync method. In this case result in CompletableFuture will contain an exception in the body.

Method	Description

Method	Description
`id`	Used to define flow identifier that will be used for report and in the trace logs
`stage`	Allows to define a `Stage` or `StageExecutor` to be executed in the flow
`flowParameter`	Allows to define a general flow parameter that will be available for all stages in the flow
`flowParameters`	Allow to define a map with flow parameters that will be available for all stages in the flow
`executionStrategy`	Allows to define flow execution strategy: `CANCEL_ON_ERROR` : on stage error – all executed stage will be cancelled by calling `cancel` method (if defined) in reverse order `IGNORE_ON_ERROR` : on stage error – the flow will be stopped by skipping execution of upstream stage and calling `onFlowError` stage (if defined)
`onFlowError`	Allows to define a `Stage` or `StageExecutor` that will be always executed if flow is failed
`onFlowCancellation`	Allows to define a `Stage` or `StageExecutor` that will be always executed if flow is cancelled
`onFlowCancellationError`	Allows to define a `Stage` or `StageExecutor` that will be always executed if flow is cancelled and one if `cancel` method is failed
`onFlowSkip`	Allows to define a `Stage` or `StageExecutor` that will be executed if flow is skipped by previous stage failure This stage is only usable for nested flow, main flow cannot be skipped

Flow Engine

A flow engine is an object that is used to execute a flow defined by a developer

Method	Description

Method	Description
`execute`	Executes flow using executor and timeout from `FlowEngineBuilder` synchronously
`executeAsync`	Executes flow using executor and timeout from `FlowEngineBuilder` asynchronously, returning `Future<Void` as a result
`getFlowStatus`	Allows to get flow status from `FlowEngine` using flow identifier Posible flow statuses: `IN_PROGRESS` `SUCCESS` `FAILED` `CANCELLED` `CANCELLATION_FAILED` `UNKNOWN` (if flow by id is not found)
`builder`	Provides a `FlowEngineBuilder` to create a `FlowEngine` object

Builder

Flow engine builder allows to customize a FlowEngine object

Method	Description

Method	Description
`name`	Allows to define flow engine name
`lastExecutionsStatusCacheSize`	Allows to define how much history preserve in memory for latest flow executions
`executor`	Allows to define custom executor for the `FlowEngine`, by default is common `ForkJoinPool` is used
`printFlowResult`	Defines if flow result must be printed in logs
`stageReportProvider`	Allows to define custom stage report provider instead of custom one

Stage Report Provider

Stage report is a useful way to customize intends and stage execution messages

Default stage report example:

var flowEngine = FlowEngine.builder()
    .printFlowResult(true)
    .executor(newSingleThreadExecutor())
    .executionTimeout(FIVE_SECONDS)
    .name("single-thread-flow-engine")
    .lastExecutionsStatusCacheSize(10)
    .build();
    
var flow = Flow.builder()
  .id("Simple Flow")
  .stage(task1)
  .stage(task2)
  .stage(task3)
  .onFlowCancellationError(onFlowCancellationErrorStage)
  .build();
  
flow.execute(flow);

17:06:51.051 INFO  FlowEngineImpl       [l-2-thread-1] Flow 'Simple Flow' execution report:
  -> Stage: task1 |> CANCELLED
  -> Stage: task2 |> CANCELLATION_FAILED with [RuntimeException]: error
  -> Stage: onFlowCancellationErrorStage |> SUCCESS

Custom stage report example:

var flowEngine = FlowEngine.builder()
    .printFlowResult(true)
    .executor(newSingleThreadExecutor())
    .executionTimeout(FIVE_SECONDS)
    .name("single-thread-flow-engine")
    .stageReportProvider(StageReportProvider.builder()
      .template("[${flowId}] -> [${stageId}] is ${statusName}")
      .errorTemplate("[${flowId}] -> [${stageId}] is ${statusName} with [${errorType}]: ${shortErrorMessage}")
      .build())
    .lastExecutionsStatusCacheSize(10)
    .build();

var flow = Flow.builder()
  .id("Simple Flow")
  .stage(task1)
  .stage(task2)
  .stage(task3)
  .onFlowCancellationError(onFlowCancellationErrorStage)
  .build();

flowEngine.execute(flow);

17:04:15.015 INFO  FlowEngineImpl       [l-2-thread-1] Flow 'Simple Flow' execution report:
  [flow-tc9s82] -> [task1] is CANCELLED
  [flow-tc9s82] -> [task2] is CANCELLATION_FAILED with [RuntimeException]: error
  [flow-tc9s82] -> [onFlowCancellationErrorStage] is SUCCESS

Stage

This interface is functional, which means that it can be used as lambda for stage definition

var stage1 = new Stage<>() {

  @Override
  public void execute(StageContext context) {
    System.out.println("Stage executed");
  }
};

var stage2 = (Stage<StageContext>) context -> System.out.println("Stage2 executed");

Flow.builder()
  .id("Simple Flow")
  .stage(stage1)
  .stage(stage2)
  .stage(context -> System.out.println("Stage3 executed"))
  .build();

The limitation of using lambdas - it’s their nature in Java and this approach won’t allow to determine in information about generic type, so context customization is not working with this case, and raw StageContext will be passed

Flow Methods

The flow engine uses flow methods to determine, what capabilities the stage provides:

Stage cancellation
Stage recovery

Method	Description

Method	Description
`execute`	Main method, which is must be defined for stage with business logic
`getId`	Returns stage identifier, which will be printed in the final report or in listenable methods
`recover`	Provides the ability to recover a stage after fail in `execute` method
`cancel`	Provides an ability to revert changed, done in `execute` and `recover` methods, if flow is failed by upstream stage
`shouldCancelIfFailed`	Defines execution of `cancel` method using `StageContext`, by default - it’s `false`

Listenable Methods

Listenable methods provided at crucial points of stage execution to provide better observability for the stage execution status

Method	Description

Method	Description
`onStart`	This method is called before stage execution
`onSuccess`	This method is called after successful stage execution including successful execution of `recover` method, if `execute` failed
`onError`	This method is called after unsuccessful stage execution (`execute` and `recover` methods)
`onCancel`	This method is called after successful execution of `cancel` method
`onCancelError`	This method is called after unsuccessful execution of `cancel` method

Parallel Stage

A parallel stage is used to specify a group of stages that need to be executed simultaneously.

var flowEngine = FlowEngine.builder().build();
var parallelStage = ParallelStage.of(task1, task2, task3);

var flow = Flow.builder()
  .id("Simple Flow")
  .stage(parallelStage)
  .build();

flowEngine.execute(flow);

Parallel Stage Builder

`id`	Used to define flow identifier that will be used for report and in the trace logs
`stage`	Allows to define a `Stage` or `StageExecutor` to be executed in the parallel
`stages`	Allows to define a list of `Stage` objects to be executed in parallel
`shouldCancelIfFailed`	Defines if parallel stage must call `cancel` method for inner stages if failed

Dynamic Stage

A dynamic stage generates a stage, flow, or parallel stage based on the data within StageContext.

var flowEngine = FlowEngine.builder().build();
var stage1 = (Stage<StageContext>) ctx -> ctx.put("values", List.of("v1", "v2"));
var parallelStage = DynamicStage.of("dynamicStageId", ctx -> {
  var flowBuilder = Flow.builder().id("Generated flow by dynamicStageId");
  var listValues = ctx.<List<String>>get("values");
  for (var value : listValues) {
    flowBuilder.stage(c -> System.out.println(value));
  }
  return flowBuilder.build();
});

var flow = Flow.builder()
  .id("Simple Flow")
  .stage(stage1)
  .stage(parallelStage)
  .build();

flowEngine.execute(flow);

Stage Context

Stage context is generated before execution of the stage and it will contain the following data

Current flow identifier
Previous stage context data
Flow parameters (nested flow parameters will always override main flow parameters until exiting the nested flow)

Stage executor

A StageExecutor is an interface that allows developers to implement a custom stage execution method. It has already been implemented as follows:

DefaultStageExecutor - used to execute all stages implementing Stage interface
FlowExecutor - used to execute all Flow objects
ParallelStageExecutor - used to execute stages in parallel
DynamicStageExecutor - used to execute DynamicStage objects

`id`	Used to define flow identifier that will be used for report and in the trace logs
`stage`	Allows to define a `Stage` or `StageExecutor` to be executed in the parallel
`stages`	Allows to define a list of `Stage` objects to be executed in parallel
`shouldCancelIfFailed`	Defines if parallel stage must call `cancel` method for inner stages if failed

Logging

flow engine provides a general logger with the name folio-flow-engine, and the log level can be customized

logger.flow-engine.name=folio-flow-engine
logger.flow-engine.level=debug

By default - all execution logs appear at the DEBUF level, and flow engine reports at the INFO level

var flowEngine = FlowEngine.builder()
    .printFlowResult(true)
    .executor(newSingleThreadExecutor())
    .executionTimeout(FIVE_SECONDS)
    .name("single-thread-flow-engine")
    .lastExecutionsStatusCacheSize(10)
    .build();
    
var flow = Flow.builder()
  .id("Simple Flow")
  .stage(task1)
  .stage(task2)
  .stage(task3)
  .onFlowCancellationError(onFlowCancellationErrorStage)
  .build();
  
flow.execute(flow);

17:06:51.051 DEBUG FlowExecutor         [l-2-thread-1] [Simple Flow] Initializing flow...
17:06:51.051 DEBUG DefaultStageExecutor [l-2-thread-1] [Simple Flow] Stage 'task1' executed with status: SUCCESS
17:06:51.051 DEBUG DefaultStageExecutor [l-2-thread-1] [Simple Flow] Stage 'task2' executed with status: FAILED
java.lang.RuntimeException: error
	at org.folio.flow.api.models.CancellableTestStage.execute(CancellableTestStage.java:11) ~[test-classes/:?]
	at org.folio.flow.impl.DefaultStageExecutor.tryExecuteStage(DefaultStageExecutor.java:129) ~[classes/:?]
	at java.base/java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:684) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:662) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:2168) ~[?:?]
	at org.folio.flow.impl.DefaultStageExecutor.execute(DefaultStageExecutor.java:95) ~[classes/:?]
	at org.folio.flow.impl.FlowExecutor.tryExecuteStageOrSkip(FlowExecutor.java:85) ~[classes/:?]
	at org.folio.flow.impl.FlowExecutor.lambda$execute$1(FlowExecutor.java:74) ~[classes/:?]
	at java.base/java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1150) [?:?]
	at java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482) [?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
	at java.base/java.lang.Thread.run(Thread.java:840) [?:?]
17:06:51.051 DEBUG FlowExecutor         [l-2-thread-1] [Simple Flow] Flow is finished with status: FAILED
17:06:51.051 DEBUG FlowExecutor         [l-2-thread-1] [Simple Flow] Initializing flow cancellation...
17:06:51.051 DEBUG DefaultStageExecutor [l-2-thread-1] [Simple Flow] Stage 'task2' cancellation failed
java.lang.RuntimeException: error
	at org.folio.flow.api.models.CancellableTestStage.cancel(CancellableTestStage.java:15) ~[test-classes/:?]
	at org.folio.flow.impl.DefaultStageExecutor.tryCancelStage(DefaultStageExecutor.java:163) ~[classes/:?]
	at java.base/java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:684) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:662) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:2168) ~[?:?]
	at org.folio.flow.impl.DefaultStageExecutor.cancel(DefaultStageExecutor.java:117) ~[classes/:?]
	at org.folio.flow.utils.StageUtils.cancelStageAsync(StageUtils.java:36) ~[classes/:?]
	at org.folio.flow.impl.FlowExecutor.lambda$cancel$8(FlowExecutor.java:116) ~[classes/:?]
	at java.base/java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1150) [?:?]
	at java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482) [?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
	at java.base/java.lang.Thread.run(Thread.java:840) [?:?]
17:06:51.051 DEBUG DefaultStageExecutor [l-2-thread-1] [Simple Flow] Stage 'task1' is cancelled
17:06:51.051 DEBUG DefaultStageExecutor [l-2-thread-1] [Simple Flow] Stage 'onFlowCancellationError' executed with status: SUCCESS
17:06:51.051 DEBUG FlowExecutor         [l-2-thread-1] [Simple Flow] Flow is cancelled with status: CANCELLATION_FAILED