diff --git a/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/SingleTaskExecution.kt b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/SingleTaskExecution.kt new file mode 100644 index 00000000..9debc1e4 --- /dev/null +++ b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/SingleTaskExecution.kt @@ -0,0 +1,28 @@ +package dev.bpmcrafters.processengineapi.impl.task + +/** + * Runs the execution of one delivered task at most once. + * The task execution is what the adapter delivery does for this task: calling the task handler and handling its failure. + * Remembers whether the task execution ran and whether it failed; a failure is rethrown unchanged. + */ +internal class SingleTaskExecution( + private val taskExecution: Runnable, + private val onRepeatedRun: () -> Unit +) : Runnable { + + private val result = lazy { runCatching { taskExecution.run() } } + + val hasRun: Boolean + get() = result.isInitialized() + + val hasFailed: Boolean + get() = hasRun && result.value.isFailure + + override fun run() { + if (hasRun) { + onRepeatedRun() + } else { + result.value.getOrThrow() + } + } +} diff --git a/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/TaskExecutionDecoration.kt b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/TaskExecutionDecoration.kt new file mode 100644 index 00000000..1dea4575 --- /dev/null +++ b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/TaskExecutionDecoration.kt @@ -0,0 +1,70 @@ +package dev.bpmcrafters.processengineapi.impl.task + +import dev.bpmcrafters.processengineapi.task.TaskExecutionDecorator +import dev.bpmcrafters.processengineapi.task.TaskInformation +import io.github.oshai.kotlinlogging.KotlinLogging +import java.util.Collections + +private val logger = KotlinLogging.logger {} + +/** + * Runs the per-task work of a delivery with the [TaskExecutionDecorator]s around it, the first one outermost. + * Adapters create one instance and call [run] around the whole per-task work, including its failure handling. + * The work always runs exactly once; a decorator that fails, skips or repeats it is only logged. + * @param decorators decorators in execution order. + * @since 1.8 + */ +class TaskExecutionDecoration(private val decorators: List) { + + /** + * Exceptions thrown by the work pass through unchanged. + * @param task information about the delivered task. + * @param payload payload delivered to the handler. + * @param execution per-task work of the delivery. + */ + fun run(task: TaskInformation, payload: Map, execution: Runnable) { + val singleTaskExecution = SingleTaskExecution(execution) { warnRepeatedRun(task) } + + try { + val decoratedExecution = decorate(readOnly(task), Collections.unmodifiableMap(payload), singleTaskExecution) + decoratedExecution.run() + } catch (e: Exception) { + if (singleTaskExecution.hasFailed) throw e + logDecoratorFailure(task, e) + } + + if (!singleTaskExecution.hasRun) { + warnSkippedRun(task) + singleTaskExecution.run() + } + } + + private fun decorate(task: TaskInformation, payload: Map, execution: Runnable): Runnable = + decorators.foldRight(execution) { decorator, inner -> + try { + decorator.decorate(task, payload, inner) + } catch (e: Exception) { + logDecoratorFailure(task, e) + inner + } + } + + private fun readOnly(task: TaskInformation) = task.copy(meta = Collections.unmodifiableMap(task.meta)) + + private fun warnRepeatedRun(task: TaskInformation) = + logger.warn { "PROCESS-ENGINE-API-018: Decorator ran task ${task.taskId} again, ignoring the repeated run." } + + private fun warnSkippedRun(task: TaskInformation) = + logger.warn { "PROCESS-ENGINE-API-017: Decorators did not run task ${task.taskId}, running it without them." } + + private fun logDecoratorFailure(task: TaskInformation, e: Exception) = + logger.error(e) { "PROCESS-ENGINE-API-016: Task execution decorator failed for task ${task.taskId}." } + + companion object { + /** + * Decoration without decorators, which just runs the work. + */ + @JvmField + val NONE = TaskExecutionDecoration(emptyList()) + } +} diff --git a/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/TaskExecutionDecorationTest.kt b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/TaskExecutionDecorationTest.kt new file mode 100644 index 00000000..d9c3ac49 --- /dev/null +++ b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/TaskExecutionDecorationTest.kt @@ -0,0 +1,122 @@ +package dev.bpmcrafters.processengineapi.impl.task + +import dev.bpmcrafters.processengineapi.task.TaskExecutionDecorator +import dev.bpmcrafters.processengineapi.task.TaskInformation +import org.assertj.core.api.Assertions.assertThat +import org.assertj.core.api.Assertions.assertThatThrownBy +import org.junit.jupiter.api.Test + +internal class TaskExecutionDecorationTest { + + private val task = TaskInformation("task-id", mutableMapOf("topicName" to "my-topic")) + private val payload = mutableMapOf("var1" to "value1") + private val recordedSteps = mutableListOf() + private val execution = Runnable { recordedSteps += "execution" } + + private fun recording(name: String) = TaskExecutionDecorator { _, _, inner -> + Runnable { + recordedSteps += "$name-before" + try { + inner.run() + } finally { + recordedSteps += "$name-after" + } + } + } + + @Test + fun `runs execution directly without decorators`() { + TaskExecutionDecoration.NONE.run(task, payload, execution) + + assertThat(recordedSteps).containsExactly("execution") + } + + @Test + fun `runs decorators around execution with the first one outermost`() { + TaskExecutionDecoration(listOf(recording("first"), recording("second"))).run(task, payload, execution) + + assertThat(recordedSteps).containsExactly("first-before", "second-before", "execution", "second-after", "first-after") + } + + @Test + fun `passes read-only task and payload to decorators`() { + var seenTask: TaskInformation? = null + var seenPayload: Map? = null + val capturing = TaskExecutionDecorator { decoratedTask, decoratedPayload, inner -> + seenTask = decoratedTask + seenPayload = decoratedPayload + inner + } + + TaskExecutionDecoration(listOf(capturing)).run(task, payload, execution) + + assertThat(seenTask).isEqualTo(task) + assertThat(seenPayload).isEqualTo(payload) + assertThatThrownBy { (seenTask!!.meta as MutableMap)["key"] = "value" } + .isInstanceOf(UnsupportedOperationException::class.java) + assertThatThrownBy { (seenPayload as MutableMap)["key"] = "value" } + .isInstanceOf(UnsupportedOperationException::class.java) + } + + @Test + fun `runs execution once if a decorator skips it`() { + val skipping = TaskExecutionDecorator { _, _, _ -> Runnable { recordedSteps += "skipping" } } + + TaskExecutionDecoration(listOf(recording("first"), skipping)).run(task, payload, execution) + + assertThat(recordedSteps).containsExactly("first-before", "skipping", "first-after", "execution") + } + + @Test + fun `runs execution once if a decorator repeats it`() { + val repeating = TaskExecutionDecorator { _, _, inner -> Runnable { inner.run(); inner.run() } } + + TaskExecutionDecoration(listOf(repeating)).run(task, payload, execution) + + assertThat(recordedSteps).containsExactly("execution") + } + + @Test + fun `skips a decorator whose decorate throws`() { + val failing = TaskExecutionDecorator { _, _, _ -> throw IllegalStateException("boom") } + + TaskExecutionDecoration(listOf(recording("first"), failing)).run(task, payload, execution) + + assertThat(recordedSteps).containsExactly("first-before", "execution", "first-after") + } + + @Test + fun `runs execution when the decorated runnable throws first`() { + val failing = TaskExecutionDecorator { _, _, _ -> Runnable { throw IllegalStateException("boom") } } + + TaskExecutionDecoration(listOf(failing)).run(task, payload, execution) + + assertThat(recordedSteps).containsExactly("execution") + } + + @Test + fun `only logs exception thrown by a decorator after the execution ran`() { + val failing = TaskExecutionDecorator { _, _, inner -> Runnable { inner.run(); throw IllegalStateException("boom") } } + + TaskExecutionDecoration(listOf(failing)).run(task, payload, execution) + + assertThat(recordedSteps).containsExactly("execution") + } + + @Test + fun `propagates exception thrown by the execution unchanged through decorators`() { + val executionError = IllegalStateException("execution") + + assertThatThrownBy { TaskExecutionDecoration(listOf(recording("first"))).run(task, payload) { throw executionError } } + .isSameAs(executionError) + assertThat(recordedSteps).containsExactly("first-before", "first-after") + } + + @Test + fun `propagates error thrown by the execution unchanged through decorators`() { + val executionError = AssertionError("execution") + + assertThatThrownBy { TaskExecutionDecoration(listOf(recording("first"))).run(task, payload) { throw executionError } } + .isSameAs(executionError) + } +} diff --git a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskExecutionDecorator.kt b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskExecutionDecorator.kt new file mode 100644 index 00000000..55bd1c3f --- /dev/null +++ b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskExecutionDecorator.kt @@ -0,0 +1,25 @@ +package dev.bpmcrafters.processengineapi.task + +/** + * Decorates the execution of an external service task, e.g. for tracing, MDC, logging or metrics. + * Similar to Spring's `TaskDecorator`, not an engine listener such as Camunda's `ExecutionListener`. + * + * `execution` is the whole per-task work of the delivery: + * the [TaskHandler] call and the delivery's failure handling. + * Code around `execution.run()` therefore also covers the delivery's failure logs. + * + * A decorator cannot change the task, as task information and payload are read-only. + * Call `execution.run()` exactly once; the adapter ensures this and only logs misbehaviour. + * Decorators run synchronously in the delivery thread, so keep them cheap. + * @since 1.8 + */ +fun interface TaskExecutionDecorator { + /** + * Decorates the execution of one task. + * @param task information about the delivered task, as an unmodifiable view. + * @param payload payload delivered to the handler, as an unmodifiable view. + * @param execution per-task work of the delivery. Run it exactly once. + * @return runnable that runs [execution] with the decoration around it. + */ + fun decorate(task: TaskInformation, payload: Map, execution: Runnable): Runnable +} diff --git a/docs/task-subscription-api.md b/docs/task-subscription-api.md index c333f402..b7fcae06 100644 --- a/docs/task-subscription-api.md +++ b/docs/task-subscription-api.md @@ -53,3 +53,89 @@ public class TaskSubscriber { ``` + +## Task execution decorators + +Cross-cutting concerns such as tracing, MDC, logging, or metrics can wrap the execution of external service tasks with a `TaskExecutionDecorator`. +It works like Spring's `TaskDecorator` and applies to `@ProcessEngineWorker` as well as to manual `subscribeForTask`. +It is not an engine listener such as a Camunda `ExecutionListener`, and it is not declared in the BPMN model. + +The adapter delivery passes its whole per-task work to the decorators: the handler call and its own failure handling. +Code before and after `execution.run()` therefore runs before and after both. +A span or MDC entry opened around `execution.run()` covers the logs of the handler, of the `process-engine-worker` and of the delivery's failure handling. + +Rules for decorators: + +- Task information and payload are read-only. Do not mutate values inside the payload, as they are shared with the handler. +- Call `execution.run()` exactly once, in the thread of the decorated runnable. If a decorator skips or repeats it, the adapter still runs it exactly once and logs a warning. +- An exception thrown by a decorator is only logged. An exception thrown by the execution passes through the decorators unchanged. +- The delivery handles a failing handler itself, so `execution.run()` returns normally even if the task failed. The failure is visible in the logs written inside the decorator's scope. + +Several decorators can be registered. They are nested in execution order, the first one outermost. +Put tracing first, so the logs of the inner decorators carry the trace id. +Decorators run synchronously in the delivery thread, so keep them cheap and non-blocking. +Release what you open in a `finally` block, otherwise a scope or MDC entry leaks into the next task on this thread. + +Decorators only take effect with an adapter that supports them. +The following example assumes an adapter starter that collects them as Spring beans ordered by `@Order`/`Ordered`. +It runs every task in its own span and puts the activity id into the MDC. + +```kotlin +@Component +@Order(0) +class TracingTaskExecutionDecorator(private val tracer: Tracer) : TaskExecutionDecorator { + + override fun decorate(task: TaskInformation, payload: Map, execution: Runnable) = Runnable { + val span = tracer.spanBuilder("${task.meta["topicName"]}.handle") + .setNoParent() + .setSpanKind(SpanKind.CONSUMER) + .setAttribute("engine.task.id", task.taskId) + .startSpan() + try { + span.makeCurrent().use { + MDC.putCloseable("activityId", task.meta[CommonRestrictions.ACTIVITY_ID]).use { + execution.run() + } + } + } finally { + span.end() + } + } +} +``` + +The same in Java, as a lambda that logs before and after each task: + +```java +@Bean +@Order(1) +TaskExecutionDecorator loggingTaskExecutionDecorator() { + return (task, payload, execution) -> () -> { + log.info("Processing task {} with variables {}", task.getTaskId(), payload); + try { + execution.run(); + } finally { + log.info("Finished task {}", task.getTaskId()); + } + }; +} +``` + +### Supporting decorators in an adapter + +An adapter creates one `TaskExecutionDecoration` (module `process-engine-api-impl`) from the decorators in execution order and passes it to its deliveries. +Each delivery of external tasks calls `run` around its whole per-task work, including the failure handling. +`TaskExecutionDecoration.NONE` just runs the work. + +```kotlin +decoration.run(taskInformation, payload) { + try { + subscription.action.accept(taskInformation, payload) + } catch (e: NotFoundException) { + logger.debug(e) { "Task ${taskInformation.taskId} is gone." } + } catch (e: Exception) { + logger.warn(e) { "Failed delivering task ${taskInformation.taskId}." } + externalTaskService.handleFailure(taskInformation.taskId, workerId, e.message, retries - 1, retryTimeout) + } +} +```