feat(embedded): decide on whether commands are executed sync or async

Introduce EngineCommandExecutor to centralise how the four core engine
calls (correlateMessage, sendSignal, startProcess, deploy) are dispatched.

The default executor submits work to ForkJoinPool.commonPool(), mirroring
the behaviour of a remote engine and keeping adapters portable across
engine implementations (embedded ↔ remote, Camunda 7 ↔ Zeebe).

Users override the Spring bean with any java.util.concurrent.Executor to
change the dispatch strategy — most commonly a same-thread executor for
@transactional coupling.

Closes #156

Author: emaarco

docs/reference-c7-embedded.md

@@ -51,3 +51,33 @@ the real type, you want to access if reading the field. For this purpose, `TaskI
 | tenantId             | String         | Tenant Id                                                                  | my_tenant                     |
 | topicName            | String         | Topic name (from BPMN) for external task                                   | topic_approve                 |
 | creationDate         | OffsetDateTime | Time stamp of task creation formatted as ISO-8601 in UTC                   | 2025-05-01T10:00:00.000Z      |
+
+## Engine Command Executor
+
+`EngineCommandExecutor` is an embedded-adapter-specific class that controls how the four core API calls
+(`correlateMessage`, `sendSignal`, `startProcess`, `deploy`) are dispatched to the embedded Camunda 7 engine.
+
+By default, all engine calls are submitted asynchronously to `ForkJoinPool.commonPool()`. This means they run on a
+**different thread** from the caller and do **not** participate in the caller's `@Transactional` context — a rollback
+on the calling thread will **not** roll back the engine operation.
+
+### Customizing execution
+
+Provide a Spring bean of type `EngineCommandExecutor` to override the default. The
+auto-configured default is annotated with `@ConditionalOnMissingBean`, so your bean takes precedence automatically.
+
+**Same-thread (synchronous) execution** — engine calls run on the calling thread and honour `@Transactional`:
+
+```kotlin
+@Bean
+fun engineCommandExecutor(): EngineCommandExecutor =
+  EngineCommandExecutor(Executor { it.run() })
+```
+
+**Virtual-thread execution** — lightweight concurrency without pinning platform threads:
+
+```kotlin
+@Bean
+fun engineCommandExecutor(): EngineCommandExecutor =
+  EngineCommandExecutor(Executors.newVirtualThreadPerTaskExecutor())
+```

engine-adapter/c7-embedded-core/src/main/kotlin/dev/bpmcrafters/processengineapi/adapter/c7/embedded/correlation/CorrelationApiImpl.kt

@@ -5,6 +5,7 @@ import dev.bpmcrafters.processengineapi.Empty
 import dev.bpmcrafters.processengineapi.MetaInfo
 import dev.bpmcrafters.processengineapi.MetaInfoAware
 import dev.bpmcrafters.processengineapi.adapter.c7.embedded.correlation.CorrelationApiImpl.Restrictions.USE_GLOBAL_CORRELATION_KEY
+import dev.bpmcrafters.processengineapi.adapter.c7.embedded.shared.EngineCommandExecutor
 import dev.bpmcrafters.processengineapi.correlation.CorrelateMessageCmd
 import dev.bpmcrafters.processengineapi.correlation.CorrelationApi
 import io.github.oshai.kotlinlogging.KotlinLogging
@@ -18,14 +19,15 @@ private val logger = KotlinLogging.logger {}
  */
 class CorrelationApiImpl(
   private val runtimeService: RuntimeService,
+  private val commandExecutor: EngineCommandExecutor,
 ) : CorrelationApi {
 
   object Restrictions {
     const val USE_GLOBAL_CORRELATION_KEY = "useGlobalCorrelationKey"
   }
 
   override fun correlateMessage(cmd: CorrelateMessageCmd): CompletableFuture<Empty> {
-    return CompletableFuture.supplyAsync {
+    return commandExecutor.execute {
       val correlation = cmd.correlation.get()
       val globalCorrelation = cmd.restrictions.containsKey(USE_GLOBAL_CORRELATION_KEY)
         && cmd.restrictions.getValue(USE_GLOBAL_CORRELATION_KEY).toBoolean()

engine-adapter/c7-embedded-core/src/main/kotlin/dev/bpmcrafters/processengineapi/adapter/c7/embedded/correlation/SignalApiImpl.kt

@@ -4,6 +4,7 @@ import dev.bpmcrafters.processengineapi.CommonRestrictions
 import dev.bpmcrafters.processengineapi.Empty
 import dev.bpmcrafters.processengineapi.MetaInfo
 import dev.bpmcrafters.processengineapi.MetaInfoAware
+import dev.bpmcrafters.processengineapi.adapter.c7.embedded.shared.EngineCommandExecutor
 import dev.bpmcrafters.processengineapi.correlation.SendSignalCmd
 import dev.bpmcrafters.processengineapi.correlation.SignalApi
 import io.github.oshai.kotlinlogging.KotlinLogging
@@ -17,12 +18,13 @@ private val logger = KotlinLogging.logger {}
  * Implementation of Signal API using local runtime service.
  */
 class SignalApiImpl(
-  private val runtimeService: RuntimeService
+  private val runtimeService: RuntimeService,
+  private val commandExecutor: EngineCommandExecutor,
 ) : SignalApi {
 
   override fun sendSignal(cmd: SendSignalCmd): CompletableFuture<Empty> {
     logger.debug { "PROCESS-ENGINE-C7-EMBEDDED-002: sending signal ${cmd.signalName}." }
-    return CompletableFuture.supplyAsync {
+    return commandExecutor.execute {
       runtimeService
         .createSignalEvent(cmd.signalName)
         .applyRestrictions(ensureSupported(cmd.restrictions))

engine-adapter/c7-embedded-core/src/main/kotlin/dev/bpmcrafters/processengineapi/adapter/c7/embedded/deploy/DeploymentApiImpl.kt

@@ -2,6 +2,7 @@ package dev.bpmcrafters.processengineapi.adapter.c7.embedded.deploy
 
 import dev.bpmcrafters.processengineapi.MetaInfo
 import dev.bpmcrafters.processengineapi.MetaInfoAware
+import dev.bpmcrafters.processengineapi.adapter.c7.embedded.shared.EngineCommandExecutor
 import dev.bpmcrafters.processengineapi.deploy.DeployBundleCommand
 import dev.bpmcrafters.processengineapi.deploy.DeploymentApi
 import dev.bpmcrafters.processengineapi.deploy.DeploymentInformation
@@ -16,13 +17,14 @@ private val logger = KotlinLogging.logger {}
  * Implementation for deployment API using repository service.
  */
 class DeploymentApiImpl(
-  private val repositoryService: RepositoryService
+  private val repositoryService: RepositoryService,
+  private val commandExecutor: EngineCommandExecutor,
 ) : DeploymentApi {
 
   override fun deploy(cmd: DeployBundleCommand): CompletableFuture<DeploymentInformation> {
     require(cmd.resources.isNotEmpty()) { "Resources must not be empty, at least one resource must be provided." }
     logger.debug { "PROCESS-ENGINE-C7-EMBEDDED-003: executing a bundle deployment with ${cmd.resources.size} resources." }
-    return CompletableFuture.supplyAsync {
+    return commandExecutor.execute {
       repositoryService
         .createDeployment()
         .apply {

engine-adapter/c7-embedded-core/src/main/kotlin/dev/bpmcrafters/processengineapi/adapter/c7/embedded/process/StartProcessApiImpl.kt

@@ -4,6 +4,7 @@ import dev.bpmcrafters.processengineapi.CommonRestrictions
 import dev.bpmcrafters.processengineapi.MetaInfo
 import dev.bpmcrafters.processengineapi.MetaInfoAware
 import dev.bpmcrafters.processengineapi.adapter.c7.embedded.correlation.applyTenantRestrictions
+import dev.bpmcrafters.processengineapi.adapter.c7.embedded.shared.EngineCommandExecutor
 import dev.bpmcrafters.processengineapi.process.*
 import io.github.oshai.kotlinlogging.KotlinLogging
 import org.camunda.bpm.engine.RepositoryService
@@ -19,30 +20,32 @@ private val logger = KotlinLogging.logger {}
 class StartProcessApiImpl(
   private val runtimeService: RuntimeService,
   private val repositoryService: RepositoryService,
+  private val commandExecutor: EngineCommandExecutor,
 ) : StartProcessApi {
 
   override fun startProcess(cmd: StartProcessCommand): CompletableFuture<ProcessInformation> {
     return when (cmd) {
       is StartProcessByDefinitionCmd ->
-        CompletableFuture.supplyAsync {
+        commandExecutor.execute {
           logger.debug { "PROCESS-ENGINE-C7-EMBEDDED-004: starting a new process instance by definition ${cmd.definitionKey}." }
           ensureSupported(cmd.restrictions)
           val payload = cmd.payloadSupplier.get()
           val tenantId = cmd.restrictions[CommonRestrictions.TENANT_ID]
           if (!tenantId.isNullOrBlank()) {
-            repositoryService
-              .createProcessDefinitionQuery()
-              .processDefinitionKey(cmd.definitionKey)
-              .tenantIdIn(tenantId)
-              .active()
-              .latestVersion()
-              .singleResult()?.let { processDefinition ->
-                runtimeService.startProcessInstanceById(
-                  processDefinition.id,
-                  payload[CommonRestrictions.BUSINESS_KEY]?.toString(),
-                  payload,
-                ).toProcessInformation()
-              }
+            val processDefinition = requireNotNull(
+              repositoryService
+                .createProcessDefinitionQuery()
+                .processDefinitionKey(cmd.definitionKey)
+                .tenantIdIn(tenantId)
+                .active()
+                .latestVersion()
+                .singleResult()
+            )
+            runtimeService.startProcessInstanceById(
+              processDefinition.id,
+              payload[CommonRestrictions.BUSINESS_KEY]?.toString(),
+              payload,
+            ).toProcessInformation()
           } else {
             runtimeService.startProcessInstanceByKey(
               cmd.definitionKey,
@@ -53,7 +56,7 @@ class StartProcessApiImpl(
         }
 
       is StartProcessByMessageCmd ->
-        CompletableFuture.supplyAsync {
+        commandExecutor.execute {
           logger.debug { "PROCESS-ENGINE-C7-EMBEDDED-005: starting a new process instance by message ${cmd.messageName}." }
           val payload = cmd.payloadSupplier.get()
           var correlationBuilder = runtimeService
@@ -69,7 +72,7 @@ class StartProcessApiImpl(
         }
 
       is StartProcessByDefinitionAtElementCmd ->
-        CompletableFuture.supplyAsync {
+        commandExecutor.execute {
           logger.debug { "PROCESS-ENGINE-C7-EMBEDDED-006: starting a new process instance by definition ${cmd.definitionKey} at element ${cmd.elementId}" }
           val startProcessCommand = StartProcessByDefinitionCmd(
             definitionKey = cmd.definitionKey,

engine-adapter/c7-embedded-core/src/main/kotlin/dev/bpmcrafters/processengineapi/adapter/c7/embedded/shared/EngineCommandExecutor.kt

@@ -0,0 +1,32 @@
+package dev.bpmcrafters.processengineapi.adapter.c7.embedded.shared
+
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.Executor
+import java.util.concurrent.ForkJoinPool
+
+/**
+ * Controls how engine commands are dispatched.
+ * By default, commands run asynchronously on [ForkJoinPool.commonPool] —
+ * mirroring the behavior of a remote engine.
+ *
+ * This design choice was made on purpose to keep adapters portable across engine implementations
+ * (embedded ↔ remote, Camunda 7 ↔ Zeebe).
+ *
+ * However, engine calls do not join the caller's transaction — a rollback triggered elsewhere
+ * will not undo what the engine already executed, risking data inconsistencies between your service and the engine.
+ *
+ * Override this bean with any [Executor] to change the dispatch strategy.
+ * The most common use case for this is transactional coupling via a same-thread executor:
+ * ```
+ * @Bean fun engineCommandExecutor() = EngineCommandExecutor { it.run() }
+ * ```
+ */
+class EngineCommandExecutor(
+  private val executor: Executor = ForkJoinPool.commonPool()
+) {
+  fun <T> execute(
+    supplier: () -> T
+  ): CompletableFuture<T> {
+    return CompletableFuture.supplyAsync({ supplier() }, executor)
+  }
+}

engine-adapter/c7-embedded-core/src/main/kotlin/dev/bpmcrafters/processengineapi/adapter/c7/embedded/testing/AbstractC7EmbeddedStage.kt

@@ -6,6 +6,7 @@ import dev.bpmcrafters.processengineapi.CommonRestrictions
 import dev.bpmcrafters.processengineapi.adapter.c7.embedded.correlation.CorrelationApiImpl
 import dev.bpmcrafters.processengineapi.adapter.c7.embedded.correlation.SignalApiImpl
 import dev.bpmcrafters.processengineapi.adapter.c7.embedded.deploy.DeploymentApiImpl
+import dev.bpmcrafters.processengineapi.adapter.c7.embedded.shared.EngineCommandExecutor
 import dev.bpmcrafters.processengineapi.adapter.c7.embedded.process.CachingProcessDefinitionMetaDataResolver
 import dev.bpmcrafters.processengineapi.adapter.c7.embedded.process.StartProcessApiImpl
 import dev.bpmcrafters.processengineapi.adapter.c7.embedded.task.completion.C7ServiceTaskCompletionApiImpl
@@ -120,12 +121,17 @@ abstract class AbstractC7EmbeddedStage<SUBTYPE : AbstractC7EmbeddedStage<SUBTYPE
     this.workerId = self().javaClass.simpleName
 
     val subscriptionRepository = InMemSubscriptionRepository()
+    val commandExecutor = EngineCommandExecutor { it.run() }
 
     startProcessApi = StartProcessApiImpl(
       runtimeService = processEngineServices.runtimeService,
       repositoryService = processEngineServices.repositoryService,
+      commandExecutor = commandExecutor,
+    )
+    deploymentApi = DeploymentApiImpl(
+      repositoryService = processEngineServices.repositoryService,
+      commandExecutor = commandExecutor,
     )
-    deploymentApi = DeploymentApiImpl(processEngineServices.repositoryService)
     userTaskCompletionApi = C7UserTaskCompletionApiImpl(processEngineServices.taskService, subscriptionRepository)
     serviceTaskCompletionApi = C7ServiceTaskCompletionApiImpl(
       workerId, processEngineServices.externalTaskService, subscriptionRepository, LinearMemoryFailureRetrySupplier(3, 3L)
@@ -152,8 +158,14 @@ abstract class AbstractC7EmbeddedStage<SUBTYPE : AbstractC7EmbeddedStage<SUBTYPE
       taskSubscriptionApi, restrictions, null, null
     )
 
-    signalApi = SignalApiImpl(processEngineServices.runtimeService)
-    correlationApi = CorrelationApiImpl(processEngineServices.runtimeService)
+    signalApi = SignalApiImpl(
+      runtimeService = processEngineServices.runtimeService,
+      commandExecutor = commandExecutor,
+    )
+    correlationApi = CorrelationApiImpl(
+      runtimeService = processEngineServices.runtimeService,
+      commandExecutor = commandExecutor,
+    )
 
     initialize()
 

engine-adapter/c7-embedded-core/src/test/kotlin/dev/bpmcrafters/processengineapi/adapter/c7/embedded/correlation/CorrelationApiImplTest.kt

@@ -1,27 +1,25 @@
 package dev.bpmcrafters.processengineapi.adapter.c7.embedded.correlation
 
 import dev.bpmcrafters.processengineapi.CommonRestrictions
+import dev.bpmcrafters.processengineapi.adapter.c7.embedded.shared.EngineCommandExecutor
 import dev.bpmcrafters.processengineapi.correlation.CorrelateMessageCmd
 import dev.bpmcrafters.processengineapi.correlation.Correlation
 import org.camunda.bpm.engine.RuntimeService
 import org.camunda.bpm.engine.runtime.MessageCorrelationBuilder
 import org.camunda.community.mockito.ProcessExpressions
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.extension.ExtendWith
-import org.mockito.InjectMocks
 import org.mockito.Mockito.mock
-import org.mockito.junit.jupiter.MockitoExtension
 import org.mockito.kotlin.verify
 import org.mockito.kotlin.verifyNoMoreInteractions
 
-@ExtendWith(MockitoExtension::class)
 class CorrelationApiImplTest {
 
   private val runtimeService: RuntimeService = mock()
-
-  @InjectMocks
-  private lateinit var correlationApi: CorrelationApiImpl
+  private val correlationApi = CorrelationApiImpl(
+    runtimeService = runtimeService,
+    commandExecutor = EngineCommandExecutor { it.run() }
+  )
 
   private lateinit var correlation: MessageCorrelationBuilder
 
@@ -64,6 +62,5 @@ class CorrelationApiImplTest {
     verify(correlation).setVariables(mapOf("some" to 1L))
     verify(correlation).processInstanceVariableEquals("myCorrelation", "varValue")
     verifyNoMoreInteractions(correlation)
-
   }
-}
+}

engine-adapter/c7-embedded-core/src/test/kotlin/dev/bpmcrafters/processengineapi/adapter/c7/embedded/correlation/SignalApiImplTest.kt

@@ -0,0 +1,54 @@
+package dev.bpmcrafters.processengineapi.adapter.c7.embedded.correlation
+
+import dev.bpmcrafters.processengineapi.Empty
+import dev.bpmcrafters.processengineapi.adapter.c7.embedded.shared.EngineCommandExecutor
+import dev.bpmcrafters.processengineapi.correlation.SendSignalCmd
+import org.assertj.core.api.Assertions.assertThat
+import org.camunda.bpm.engine.RuntimeService
+import org.camunda.bpm.engine.runtime.SignalEventReceivedBuilder
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
+import org.mockito.Mock
+import org.mockito.junit.jupiter.MockitoExtension
+import org.mockito.kotlin.any
+import org.mockito.kotlin.doAnswer
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.verify
+import org.mockito.kotlin.verifyNoMoreInteractions
+import org.mockito.kotlin.whenever
+
+@ExtendWith(MockitoExtension::class)
+class SignalApiImplTest {
+
+  @Mock
+  private lateinit var runtimeService: RuntimeService
+
+  private lateinit var signalApi: SignalApiImpl
+
+  @BeforeEach
+  fun setUp() {
+    signalApi = SignalApiImpl(
+      runtimeService = runtimeService,
+      commandExecutor = EngineCommandExecutor { it.run() }
+    )
+  }
+
+  @Test
+  fun `should send signal and return completedFuture`() {
+    val signalBuilder = mock<SignalEventReceivedBuilder>()
+    val payload = mapOf("key" to "value")
+    val cmd = SendSignalCmd(signalName = "mySignal", payloadSupplier = { payload })
+    whenever(runtimeService.createSignalEvent(any())).thenReturn(signalBuilder)
+    whenever(signalBuilder.setVariables(any())).thenReturn(signalBuilder)
+    doAnswer { }.whenever(signalBuilder).send()
+
+    val future = signalApi.sendSignal(cmd = cmd).get()
+
+    assertThat(future).isEqualTo(Empty)
+    verify(runtimeService).createSignalEvent("mySignal")
+    verify(signalBuilder).setVariables(payload)
+    verify(signalBuilder).send()
+    verifyNoMoreInteractions(signalBuilder)
+  }
+}

engine-adapter/c7-embedded-core/src/test/kotlin/dev/bpmcrafters/processengineapi/adapter/c7/embedded/deploy/DeploymentApiImplTest.kt

@@ -1,5 +1,6 @@
 package dev.bpmcrafters.processengineapi.adapter.c7.embedded.deploy
 
+import dev.bpmcrafters.processengineapi.adapter.c7.embedded.shared.EngineCommandExecutor
 import dev.bpmcrafters.processengineapi.deploy.DeployBundleCommand
 import dev.bpmcrafters.processengineapi.deploy.NamedResource
 import org.assertj.core.api.Assertions.assertThat
@@ -17,7 +18,10 @@ import java.util.*
 class DeploymentApiImplTest {
 
   private val repositoryService: RepositoryService = mock()
-  private val deploymentApiImpl = DeploymentApiImpl(repositoryService)
+  private val deploymentApiImpl = DeploymentApiImpl(
+    repositoryService = repositoryService,
+    commandExecutor = EngineCommandExecutor { it.run() }
+  )
 
   @Test
   fun `empty tenant id was not set`() {