This is an automated email from the ASF dual-hosted git repository.
martinweiler pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-kie.git
The following commit(s) were added to refs/heads/main by this push:
new ff5db7b047c [kie-issues#7162] Delay ProcessInstance lock release to
transaction commit to avoid stale read in multi-threaded scenarios (#7163)
ff5db7b047c is described below
commit ff5db7b047c7daacd29382939ae2028a210a7bc7
Author: Martin Weiler <[email protected]>
AuthorDate: Fri Oct 9 08:58:59 2026 -0600
[kie-issues#7162] Delay ProcessInstance lock release to transaction commit
to avoid stale read in multi-threaded scenarios (#7163)
* [kie-issues#7162] Delay ProcessInstance lock release to transaction
commit to avoid stale read in multi-threaded scenarios
* Simplify the approach by moving the transaction related code directly
into ProcessInstanceAtomicLockStrategy
* Restore isReentrant check location
* - make VertxJobScheduler user the vertx platform integration
* - api cleanup
* Align Spring Boot unlock logic with Quarkus using afterCompletion instead
of afterCommit
* Make Vertx injection optional in SpringbootJobsService
---------
Co-authored-by: pefernan <[email protected]>
---
.../persistence/jdbc/JDBCProcessInstances.java | 29 ++-
.../app/jobs/quarkus/QuarkusJobsService.java | 5 +
.../app/jobs/springboot/SpringbootJobsService.java | 6 +
.../TransactionJobTimeoutInterceptor.java | 18 +-
.../workflow/core/node/AsyncEventNodeInstance.java | 3 +-
.../process/impl/AbstractProcessInstance.java | 41 +++-
.../ContextAwareProcessInstanceLockStrategy.java | 15 ++
.../lock/ProcessInstanceAtomicLockStrategy.java | 94 ++++++++-
.../impl/lock/ProcessInstanceLockStrategy.java | 31 ++-
.../kogito/app/jobs/api/JobSchedulerBuilder.java | 10 +-
.../kogito/app/jobs/impl/VertxJobScheduler.java | 23 ++-
.../integrations/ProcessInstanceJobExecutor.java | 5 +-
.../app/jobs/impl/VertxJobSchedulerTest.java | 47 +++++
.../quarkus/JDBCProcessInstancesFactory.java | 31 +++
.../org/kie/kogito/BranchExecutionCounter.java | 49 +++++
...er.java => BranchExecutionCounterResource.java} | 28 ++-
.../kogito/wih/BranchCounterWorkItemHandler.java | 63 ++++++
.../main/java/org/kie/kogito/wih/WIHRegister.java | 1 +
.../src/main/resources/asyncsplit.bpmn | 223 +++++++++++++++++++++
.../org/kie/kogito/it/OptimisticLockingTest.java | 54 +++++
.../pom.xml | 8 +
.../pom.xml | 17 ++
.../springboot/JDBCProcessInstancesFactory.java | 22 ++
.../org/kie/kogito/process/KogitoBeanProducer.java | 8 +-
24 files changed, 793 insertions(+), 38 deletions(-)
diff --git
a/kogito-addons/common/persistence/jdbc/src/main/java/org/kie/kogito/persistence/jdbc/JDBCProcessInstances.java
b/kogito-addons/common/persistence/jdbc/src/main/java/org/kie/kogito/persistence/jdbc/JDBCProcessInstances.java
index 414ac08c2a2..205a83e99fa 100644
---
a/kogito-addons/common/persistence/jdbc/src/main/java/org/kie/kogito/persistence/jdbc/JDBCProcessInstances.java
+++
b/kogito-addons/common/persistence/jdbc/src/main/java/org/kie/kogito/persistence/jdbc/JDBCProcessInstances.java
@@ -21,6 +21,7 @@ package org.kie.kogito.persistence.jdbc;
import java.util.HashSet;
import java.util.Optional;
import java.util.UUID;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -49,6 +50,14 @@ public class JDBCProcessInstances<T extends Model>
implements MutableProcessInst
private final boolean lock;
private final Repository repository;
+ /**
+ * Optional transaction registrar: when set, {@link #connectInstance}
passes it to the
+ * process instance so write operations release the JVM lock only after
the surrounding
+ * transaction commits. Null means "unlock immediately" — safe for
non-transactional or
+ * test environments.
+ */
+ private Consumer<Runnable> transactionRegistrar;
+
public JDBCProcessInstances(Process<?> process, DataSource dataSource,
boolean lock) {
this(process, dataSource, lock, null, null);
}
@@ -65,6 +74,16 @@ public class JDBCProcessInstances<T extends Model>
implements MutableProcessInst
this.repository = new GenericRepository(dataSource, processes);
}
+ /**
+ * Sets the transaction registrar to enable deferred lock release.
+ * Called by the framework-specific factory (Quarkus/Spring) when a
transaction manager is
+ * available. The consumer receives an unlock {@link Runnable} and must
arrange for it to
+ * run after the surrounding transaction commits (or immediately when no
transaction is active).
+ */
+ public void setTransactionRegistrar(Consumer<Runnable>
transactionRegistrar) {
+ this.transactionRegistrar = transactionRegistrar;
+ }
+
@Override
public boolean exists(String id) {
return findById(id).isPresent();
@@ -96,7 +115,9 @@ public class JDBCProcessInstances<T extends Model>
implements MutableProcessInst
if (isActive(instance) || instance.status() ==
ProcessInstance.STATE_PENDING) {
String[] eventTypes = getUniqueEvents(instance);
if (lock) {
- boolean isUpdated = repository.updateWithLock(process.id(),
process.version(), UUID.fromString(id),
marshaller.marshallProcessInstance(instance), instance.version(), eventTypes);
+ long versionBeforeWrite = instance.version();
+ boolean isUpdated = repository.updateWithLock(process.id(),
process.version(), UUID.fromString(id),
marshaller.marshallProcessInstance(instance), versionBeforeWrite, eventTypes);
+ LOGGER.trace("updateWithLock [{}] UPDATE WHERE version={} ->
isUpdated={}", id, versionBeforeWrite, isUpdated);
if (!isUpdated) {
throw new ProcessInstanceOptimisticLockingException(id);
}
@@ -168,12 +189,16 @@ public class JDBCProcessInstances<T extends Model>
implements MutableProcessInst
}
private void connectInstance(ProcessInstance<?> instance) {
- ((AbstractProcessInstance<?>) instance).internalSetReloadSupplier(pi
-> {
+ AbstractProcessInstance<?> api = (AbstractProcessInstance<?>) instance;
+ api.internalSetReloadSupplier(pi -> {
Repository.Record r = repository.findByIdInternal(process.id(),
process.version(), UUID.fromString(pi.id())).orElseThrow();
pi.setVersion(r.version());
marshaller.createdReloadFunction(r::payload).accept(pi);
pi.internalGetProcessInstance().setRootProcessId(r.rootProcessId());
pi.internalGetProcessInstance().setRootProcessVersion(r.rootProcessVersion());
});
+ if (transactionRegistrar != null) {
+ api.internalSetTransactionRegistrar(transactionRegistrar);
+ }
}
}
diff --git
a/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/quarkus/QuarkusJobsService.java
b/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/quarkus/QuarkusJobsService.java
index 8013800618c..efec1b8b687 100644
---
a/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/quarkus/QuarkusJobsService.java
+++
b/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/quarkus/QuarkusJobsService.java
@@ -38,6 +38,7 @@ import org.kie.kogito.jobs.JobDescription;
import org.kie.kogito.jobs.JobsService;
import io.quarkus.runtime.Startup;
+import io.vertx.core.Vertx;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
@@ -98,6 +99,9 @@ public class QuarkusJobsService implements JobsService {
@Inject
TransactionRollbackMarker transactionRollbackMarker;
+ @Inject
+ Vertx vertx;
+
@PostConstruct
public void init() {
this.jobScheduler = JobSchedulerBuilder.newJobSchedulerBuilder()
@@ -117,6 +121,7 @@ public class QuarkusJobsService implements JobsService {
.withTimeoutInterceptor(
new TransactionJobTimeoutInterceptor(),
new
ErrorHandlingJobTimeoutInterceptor(exceptionHandlers.stream().toList()))
+ .withVertx(vertx)
.withExceptionDetailsExtractor(exceptionDetailsExtractor)
.withTransactionRollbackMarker(transactionRollbackMarker)
.withNumberOfWorkerThreads(numberOfWorkerThreads)
diff --git
a/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/SpringbootJobsService.java
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/SpringbootJobsService.java
index b71251990fc..4d109d09176 100644
---
a/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/SpringbootJobsService.java
+++
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/SpringbootJobsService.java
@@ -46,6 +46,8 @@ import
org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import
org.springframework.transaction.support.TransactionSynchronizationManager;
+import io.vertx.core.Vertx;
+
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
@@ -97,6 +99,9 @@ public class SpringbootJobsService implements JobsService {
@Autowired
protected TransactionRollbackMarker transactionRollbackMarker;
+ @Autowired(required = false)
+ protected Vertx vertx;
+
@PostConstruct
public void init() {
this.jobScheduler = JobSchedulerBuilder.newJobSchedulerBuilder()
@@ -116,6 +121,7 @@ public class SpringbootJobsService implements JobsService {
.withTimeoutInterceptor(
new
ErrorHandlingJobTimeoutInterceptor(ofNullable(exceptionHandlers).stream().toList()),
new
TransactionJobTimeoutInterceptor(transactionManager))
+ .withVertx(vertx)
.withExceptionDetailsExtractor(exceptionDetailsExtractor)
.withTransactionRollbackMarker(transactionRollbackMarker)
.withNumberOfWorkerThreads(numberOfWorkerThreads)
diff --git
a/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/TransactionJobTimeoutInterceptor.java
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/TransactionJobTimeoutInterceptor.java
index f2ef72f0839..70d708d9a83 100644
---
a/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/TransactionJobTimeoutInterceptor.java
+++
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/TransactionJobTimeoutInterceptor.java
@@ -23,15 +23,16 @@ import java.util.concurrent.Callable;
import org.kie.kogito.app.jobs.api.JobTimeoutExecution;
import org.kie.kogito.app.jobs.api.JobTimeoutInterceptor;
import org.springframework.transaction.PlatformTransactionManager;
-import
org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.TransactionTemplate;
public class TransactionJobTimeoutInterceptor implements JobTimeoutInterceptor
{
- private TransactionTemplate transactionTemplate;
+ private final TransactionTemplate transactionTemplate;
public TransactionJobTimeoutInterceptor(PlatformTransactionManager
transactionManager) {
this.transactionTemplate = new TransactionTemplate(transactionManager);
+
this.transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
}
@Override
@@ -41,25 +42,14 @@ public class TransactionJobTimeoutInterceptor implements
JobTimeoutInterceptor {
@Override
public JobTimeoutExecution call() throws Exception {
return transactionTemplate.execute(status -> {
+
SpringBootTransactionRollbackMarker.setCurrentTransactionStatus(status);
try {
- // Initialize transaction synchronization if not
already active
- // This is needed for
TransactionSynchronizationManager callbacks to work
- if
(!TransactionSynchronizationManager.isSynchronizationActive()) {
-
TransactionSynchronizationManager.initSynchronization();
- }
-
- // Store the transaction status in thread-local for
SpringBootTransactionRollbackMarker
-
SpringBootTransactionRollbackMarker.setCurrentTransactionStatus(status);
-
return callable.call();
} catch (RuntimeException e) {
- // Re-throw RuntimeException to trigger rollback
throw e;
} catch (Exception e) {
- // Wrap checked exceptions in RuntimeException to
trigger rollback
throw new RuntimeException(e);
} finally {
- // Clear the transaction status from thread-local
SpringBootTransactionRollbackMarker.clearCurrentTransactionStatus();
}
});
diff --git
a/kogito-jbpm/jbpm-flow/src/main/java/org/jbpm/workflow/core/node/AsyncEventNodeInstance.java
b/kogito-jbpm/jbpm-flow/src/main/java/org/jbpm/workflow/core/node/AsyncEventNodeInstance.java
index bd82b536af5..a4d7d624e34 100644
---
a/kogito-jbpm/jbpm-flow/src/main/java/org/jbpm/workflow/core/node/AsyncEventNodeInstance.java
+++
b/kogito-jbpm/jbpm-flow/src/main/java/org/jbpm/workflow/core/node/AsyncEventNodeInstance.java
@@ -22,7 +22,6 @@ import java.time.ZonedDateTime;
import java.time.temporal.ChronoUnit;
import java.util.Collection;
import java.util.Objects;
-import java.util.Optional;
import org.jbpm.process.instance.InternalProcessRuntime;
import org.jbpm.workflow.instance.NodeInstanceContainer;
@@ -112,7 +111,7 @@ public class AsyncEventNodeInstance extends
EventNodeInstance {
.processVersion(instance.getProcessInstance().getProcessVersion())
.rootProcessId(instance.getProcessInstance().getRootProcessId())
.rootProcessVersion(instance.getProcessInstance().getRootProcessVersion())
-
.nodeInstanceId(Optional.ofNullable(from).map(KogitoNodeInstance::getStringId).orElse(null))
+ .nodeInstanceId(this.getStringId())
.build();
JobsService jobService = processRuntime.getJobsService();
String jobId = jobService.scheduleJob(jobDescription);
diff --git
a/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/AbstractProcessInstance.java
b/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/AbstractProcessInstance.java
index 842fdb08ffc..e7925c88cab 100644
---
a/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/AbstractProcessInstance.java
+++
b/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/AbstractProcessInstance.java
@@ -111,6 +111,23 @@ public abstract class AbstractProcessInstance<T extends
Model> implements Proces
private ProcessInstanceLockStrategy processInstanceLockStrategy;
+ /**
+ * Transaction registrar for deferred lock release. When non-null (set by
+ * {@link JDBCProcessInstances#connectInstance} via {@link
#internalSetTransactionRegistrar}),
+ * write operations defer the JVM-level lock release until after the
surrounding transaction
+ * commits, preventing other threads from reading a stale optimistic-lock
version.
+ */
+ private Consumer<Runnable> transactionRegistrar;
+
+ /**
+ * Sets the transaction registrar so that write operations release the JVM
lock only after
+ * the surrounding transaction commits. Called by the persistence layer
after the instance
+ * is unmarshalled from the DB. A {@code null} value (the default) means
unlock immediately.
+ */
+ public void internalSetTransactionRegistrar(Consumer<Runnable>
transactionRegistrar) {
+ this.transactionRegistrar = transactionRegistrar;
+ }
+
public AbstractProcessInstance(AbstractProcess<T> process, T variables,
ProcessRuntime rt) {
this(process, variables, null, rt);
}
@@ -612,11 +629,26 @@ public abstract class AbstractProcessInstance<T extends
Model> implements Proces
}
private <R> R
executeInWorkflowProcessInstance(Function<WorkflowProcessInstanceImpl, R>
execution) {
- // Check if this is a reentrant call before entering the lock
+ // Check reentrance before acquiring the lock: isLockedByCurrentThread
returns true only
+ // when the current thread *already* holds the lock, so it must be
read before
+ // executeWriteOperation acquires it — otherwise every outermost call
looks reentrant.
boolean isReentrant =
processInstanceLockStrategy.isLockedByCurrentThread(id);
-
- return processInstanceLockStrategy.executeOperation(id, () -> {
+ return processInstanceLockStrategy.executeWriteOperation(id, () -> {
+ // The process instance (and its optimistic-lock version) may have
been loaded from
+ // the DB *before* this lock was acquired (e.g. by
JDBCProcessInstances.findById in
+ // ProcessInstanceJobExecutor). Discard that stale snapshot so
that the SELECT issued
+ // by reloadSupplier happens *inside* the lock, guaranteeing the
version we write back
+ // was read under mutual exclusion with every other writer.
+ // Guards:
+ // !isReentrant — reentrant callers must not null out the
parent's live state.
+ // reloadSupplier != null — brand-new instances (start path)
have no supplier yet;
+ // their processInstance was set in the
constructor.
+ if (!isReentrant && reloadSupplier != null) {
+ LOG.trace("executeInWorkflowProcessInstance [{}] discarding
stale snapshot (version={}) before lock-guarded reload", id, this.version);
+ this.processInstance = null;
+ }
WorkflowProcessInstanceImpl workflowProcessInstance =
internalLoadProcessInstanceState();
+ LOG.trace("executeInWorkflowProcessInstance [{}] state loaded,
version={}, reentrant={}", id, this.version, isReentrant);
if (isProcessInstanceConnected()) {
getProcessRuntime().getProcessInstanceManager().addProcessInstance(workflowProcessInstance);
}
@@ -636,6 +668,7 @@ public abstract class AbstractProcessInstance<T extends
Model> implements Proces
}
if (isProcessInstanceConnected()) {
+ LOG.trace("executeInWorkflowProcessInstance [{}] about to
syncPersistence with version={}", id, this.version);
syncPersistence(workflowProcessInstance);
getProcessRuntime().getProcessInstanceManager().removeProcessInstance(workflowProcessInstance);
}
@@ -646,7 +679,7 @@ public abstract class AbstractProcessInstance<T extends
Model> implements Proces
}
return outcome;
- });
+ }, transactionRegistrar);
}
@Override
diff --git
a/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ContextAwareProcessInstanceLockStrategy.java
b/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ContextAwareProcessInstanceLockStrategy.java
index eabf6a15e13..92bd63a7782 100644
---
a/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ContextAwareProcessInstanceLockStrategy.java
+++
b/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ContextAwareProcessInstanceLockStrategy.java
@@ -18,6 +18,8 @@
*/
package org.kie.kogito.process.impl.lock;
+import java.util.function.Consumer;
+
import org.kie.kogito.services.context.ProcessInstanceContext;
/**
@@ -67,6 +69,19 @@ public class ContextAwareProcessInstanceLockStrategy
implements ProcessInstanceL
});
}
+ @Override
+ public <T> T executeWriteOperation(String processInstanceId,
WorkflowAtomicExecutor<T> operation,
+ Consumer<Runnable> transactionRegistrar) {
+ return delegate.executeWriteOperation(processInstanceId, () -> {
+ ProcessInstanceContext.setProcessInstanceId(processInstanceId);
+ try {
+ return operation.execute();
+ } finally {
+ ProcessInstanceContext.clear();
+ }
+ }, transactionRegistrar);
+ }
+
@Override
public boolean isLockedByCurrentThread(String processInstanceId) {
return delegate.isLockedByCurrentThread(processInstanceId);
diff --git
a/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ProcessInstanceAtomicLockStrategy.java
b/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ProcessInstanceAtomicLockStrategy.java
index 0cd3a275317..4cf9b1847fc 100644
---
a/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ProcessInstanceAtomicLockStrategy.java
+++
b/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ProcessInstanceAtomicLockStrategy.java
@@ -21,6 +21,7 @@ package org.kie.kogito.process.impl.lock;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -101,7 +102,7 @@ public class ProcessInstanceAtomicLockStrategy implements
ProcessInstanceLockStr
LOG.trace("Lock released for {}", processInstanceId);
}
- // evaluate atomically if the lock is still in used before
removing it.
+ // evaluate atomically if the lock is still in use before removing
it.
locks.computeIfPresent(processInstanceId, (pid, holder) -> {
holder.removeReference();
if (holder.isReferenced()) {
@@ -112,7 +113,96 @@ public class ProcessInstanceAtomicLockStrategy implements
ProcessInstanceLockStr
}
});
}
+ }
+
+ /**
+ * Write path with optional deferred unlock.
+ *
+ * <p>
+ * When {@code transactionRegistrar} is non-null and the current thread
does not already
+ * hold the lock (non-reentrant call), the lock release is deferred to
after the surrounding
+ * transaction commits. The registrar receives an unlock {@link Runnable}
and must arrange
+ * for it to run in an {@code afterCompletion}/{@code afterCommit}
callback; when no active
+ * transaction is present the registrar must run the action immediately.
+ *
+ * <p>
+ * Reentrant calls (the lock is already held by the current thread) always
release
+ * immediately in the {@code finally} block, regardless of {@code
transactionRegistrar},
+ * because the outermost non-reentrant frame is responsible for the
deferred release.
+ *
+ * <p>
+ * On the exception path the lock is always released immediately so that
other threads
+ * are not blocked behind a failed operation.
+ */
+ @Override
+ public <T> T executeWriteOperation(String processInstanceId,
WorkflowAtomicExecutor<T> executor,
+ Consumer<Runnable> transactionRegistrar) {
+ if (transactionRegistrar == null) {
+ return executeOperation(processInstanceId, executor);
+ }
+
+ boolean alreadyHeld = isLockedByCurrentThread(processInstanceId);
+
+ ProcessInstanceLockHolder holder = locks.compute(processInstanceId,
(pid, h) -> {
+ ProcessInstanceLockHolder newHolder = h == null ? new
ProcessInstanceLockHolder() : h;
+ newHolder.addReference();
+ LOG.trace("Creating lock {} from list as none is waiting for it by
{}", newHolder.lock, pid);
+ return newHolder;
+ });
+ boolean alreadyAcquired = holder.isHeldByCurrentThread();
+ boolean deferred = false;
+ try {
+ if (!alreadyAcquired) {
+ LOG.trace("About to acquire lock for {}", processInstanceId);
+ }
+ holder.lock();
+ if (!alreadyAcquired) {
+ LOG.trace("Lock acquired for {}", processInstanceId);
+ }
+
+ T result = executor.execute();
+
+ if (!alreadyHeld) {
+ // Defer lock release to after the surrounding transaction
commits.
+ // The registrar captures the holder reference so the
post-commit callback
+ // can unlock it even after this stack frame has returned.
+ final ProcessInstanceLockHolder deferredHolder = holder;
+ LOG.trace("Deferring lock release to post-commit for {}",
processInstanceId);
+ transactionRegistrar.accept(() -> {
+ deferredHolder.unlock();
+ LOG.trace("Lock released (post-commit) for {}",
processInstanceId);
+ locks.computeIfPresent(processInstanceId, (pid, h) -> {
+ h.removeReference();
+ if (h.isReferenced()) {
+ return h;
+ } else {
+ LOG.trace("Removing lock {} from list as none is
waiting for it by {}", h.lock, pid);
+ return null;
+ }
+ });
+ });
+ deferred = true;
+ }
+ return result;
+ } finally {
+ if (!deferred) {
+ // Reentrant call, or exception path: release immediately.
+ holder.unlock();
+ if (!alreadyAcquired) {
+ LOG.trace("Lock released for {}", processInstanceId);
+ }
+ locks.computeIfPresent(processInstanceId, (pid, h) -> {
+ h.removeReference();
+ if (h.isReferenced()) {
+ return h;
+ } else {
+ LOG.trace("Removing lock {} from list as none is
waiting for it by {}", h.lock, pid);
+ return null;
+ }
+ });
+ }
+ }
}
@Override
@@ -121,7 +211,7 @@ public class ProcessInstanceAtomicLockStrategy implements
ProcessInstanceLockStr
return holder != null && holder.isHeldByCurrentThread();
}
- public static synchronized ProcessInstanceLockStrategy instance() {
+ public static synchronized ProcessInstanceAtomicLockStrategy instance() {
if (INSTANCE == null) {
INSTANCE = new ProcessInstanceAtomicLockStrategy();
}
diff --git
a/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ProcessInstanceLockStrategy.java
b/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ProcessInstanceLockStrategy.java
index 5d0a27d5aaa..a9e3605f307 100644
---
a/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ProcessInstanceLockStrategy.java
+++
b/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/lock/ProcessInstanceLockStrategy.java
@@ -18,9 +18,38 @@
*/
package org.kie.kogito.process.impl.lock;
+import java.util.function.Consumer;
+
public interface ProcessInstanceLockStrategy {
- <T> T executeOperation(String processInstanceId, WorkflowAtomicExecutor<T>
empty);
+ <T> T executeOperation(String processInstanceId, WorkflowAtomicExecutor<T>
executor);
+
+ /**
+ * Executes a write operation under the lock, optionally deferring the
lock release to after
+ * the surrounding transaction commits.
+ *
+ * <p>
+ * When {@code transactionRegistrar} is non-null, implementations should
arrange for the
+ * lock to be released only after the surrounding transaction commits, by
passing an unlock
+ * {@link Runnable} to the registrar. The registrar must run the action
immediately when no
+ * active transaction is present, so this is safe in non-transactional
environments too.
+ *
+ * <p>
+ * The default implementation ignores {@code transactionRegistrar} and
delegates to
+ * {@link #executeOperation}, making it safe for non-transactional or test
environments.
+ *
+ * @param processInstanceId the process instance id
+ * @param executor the operation to execute
+ * @param transactionRegistrar accepts an unlock {@link Runnable} and
arranges for it to run
+ * after the surrounding transaction commits; {@code null} means
+ * unlock immediately in the {@code finally} block
+ * @param <T> the return type
+ * @return the result of the operation
+ */
+ default <T> T executeWriteOperation(String processInstanceId,
WorkflowAtomicExecutor<T> executor,
+ Consumer<Runnable> transactionRegistrar) {
+ return executeOperation(processInstanceId, executor);
+ }
/**
* Checks if the current thread already holds the lock for the given
process instance.
diff --git
a/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/api/JobSchedulerBuilder.java
b/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/api/JobSchedulerBuilder.java
index 8376af6256e..5432cc09357 100644
---
a/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/api/JobSchedulerBuilder.java
+++
b/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/api/JobSchedulerBuilder.java
@@ -25,10 +25,12 @@ import org.kie.kogito.app.jobs.spi.JobStore;
import org.kie.kogito.app.jobs.spi.TransactionRollbackMarker;
import org.kie.kogito.event.EventPublisher;
+import io.vertx.core.Vertx;
+
public interface JobSchedulerBuilder {
static JobSchedulerBuilder newJobSchedulerBuilder() {
- return new VertxJobScheduler().new VertxJobSchedulerBuilder();
+ return VertxJobScheduler.builder();
}
JobScheduler build();
@@ -57,6 +59,12 @@ public interface JobSchedulerBuilder {
JobSchedulerBuilder withNumberOfWorkerThreads(Integer
numberOfWorkerThreads);
+ /**
+ * Supplies the platform-managed Vert.x instance to use as the execution
runtime.
+ * If not called, the scheduler creates and owns a standalone instance.
+ */
+ JobSchedulerBuilder withVertx(Vertx vertx);
+
JobSchedulerBuilder withJobSynchronization(JobSynchronization
jobSynchronization);
JobSchedulerBuilder withJobDescriptorMergers(JobDescriptionMerger...
jobDescriptionMergers);
diff --git
a/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java
b/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java
index 8a5f284acce..0b3bcb9bfcc 100644
---
a/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java
+++
b/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java
@@ -91,6 +91,8 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
private Vertx vertx;
+ private boolean ownsVertx;
+
private WorkerExecutor workerExecutor;
private JobContextFactory jobContextFactory;
@@ -205,6 +207,12 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
return this;
}
+ @Override
+ public JobSchedulerBuilder withVertx(Vertx vertx) {
+ VertxJobScheduler.this.vertx = vertx;
+ return this;
+ }
+
@Override
public JobSchedulerBuilder
withJobDescriptorMergers(JobDescriptionMerger... jobDescriptionMergers) {
VertxJobScheduler.this.jobDescriptionMergers.addAll(List.of(jobDescriptionMergers));
@@ -225,6 +233,10 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
}
+ public static VertxJobSchedulerBuilder builder() {
+ return new VertxJobScheduler().new VertxJobSchedulerBuilder();
+ }
+
public VertxJobScheduler() {
this.jobExecutors = new ArrayList<>();
this.jobStore = new MemoryJobStore();
@@ -326,7 +338,11 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
@Override
public void init() {
- this.vertx = Vertx.builder().build();
+ if (this.vertx == null) {
+ LOG.info("No Vert.x instance provided — creating a standalone one
with default settings.");
+ this.vertx = Vertx.builder().build();
+ this.ownsVertx = true;
+ }
this.workerExecutor = this.vertx.createSharedWorkerExecutor("Jobs",
numberOfWorkerThreads);
this.maxRefreshJobsIntervalWindow =
Math.max(maxRefreshJobsIntervalWindow, refreshJobsInterval);
this.refreshJobsIntervalTimerId = this.vertx.setPeriodic(0L,
refreshJobsInterval, this);
@@ -352,12 +368,15 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
// clean up
this.workerExecutor.close();
- this.vertx.close();
+ if (this.ownsVertx) {
+ this.vertx.close();
+ }
this.jobsScheduled.clear();
this.refreshJobsIntervalTimerId = null;
this.workerExecutor = null;
this.vertx = null;
+ this.ownsVertx = false;
}
@Override
diff --git
a/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/integrations/ProcessInstanceJobExecutor.java
b/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/integrations/ProcessInstanceJobExecutor.java
index c2367b285cc..21ed9fc6141 100644
---
a/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/integrations/ProcessInstanceJobExecutor.java
+++
b/kogito-jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/integrations/ProcessInstanceJobExecutor.java
@@ -32,9 +32,8 @@ import org.kie.kogito.uow.UnitOfWorkManager;
public class ProcessInstanceJobExecutor implements JobExecutor {
public static final String SIGNAL = "timerTriggered";
- private Processes processes;
-
- private UnitOfWorkManager uom;
+ private final Processes processes;
+ private final UnitOfWorkManager uom;
public ProcessInstanceJobExecutor(Processes processes, UnitOfWorkManager
unitOfWorkManager) {
this.processes = processes;
diff --git
a/kogito-jobs/jobs-common-embedded/src/test/java/org/kie/kogito/app/jobs/impl/VertxJobSchedulerTest.java
b/kogito-jobs/jobs-common-embedded/src/test/java/org/kie/kogito/app/jobs/impl/VertxJobSchedulerTest.java
index 17afb1e2892..d8ccb384b5b 100644
---
a/kogito-jobs/jobs-common-embedded/src/test/java/org/kie/kogito/app/jobs/impl/VertxJobSchedulerTest.java
+++
b/kogito-jobs/jobs-common-embedded/src/test/java/org/kie/kogito/app/jobs/impl/VertxJobSchedulerTest.java
@@ -40,10 +40,57 @@ import org.kie.kogito.jobs.service.model.JobStatus;
import org.kie.kogito.jobs.service.model.ScheduledJob;
import org.kie.kogito.timer.impl.SimpleTimerTrigger;
+import io.vertx.core.Vertx;
+
import static org.assertj.core.api.Assertions.assertThat;
public class VertxJobSchedulerTest {
+ /**
+ * Verifies that when a platform-managed Vert.x instance is supplied via
+ * {@link JobSchedulerBuilder#withVertx(Vertx)}, the scheduler uses it
instead of
+ * creating a standalone instance. The supplied Vert.x is closed by the
caller, not
+ * by the scheduler — verified by asserting the scheduler completes
normally and the
+ * external instance is still running after {@code jobScheduler.close()}.
+ */
+ @Test
+ public void testWithExternalVertxUsesSuppliedInstance() throws Exception {
+ final String jobId = "external-vertx-test";
+ Vertx externalVertx = Vertx.vertx();
+ try {
+ LatchExecutionJobSchedulerListener listener = new
LatchExecutionJobSchedulerListener();
+ TestJobExecutor executor = new TestJobExecutor();
+ JobStore jobStore = new MemoryJobStore();
+ JobContextFactory jobContextFactory = new
MemoryJobContextFactory();
+
+ JobScheduler jobScheduler =
JobSchedulerBuilder.newJobSchedulerBuilder()
+ .withJobExecutors(executor)
+ .withJobEventAdapters(new TestJobDetailsEventAdapter())
+ .withEventPublishers(new TestEventPublisher())
+ .withJobContextFactory(jobContextFactory)
+ .withJobStore(jobStore)
+ .withJobSchedulerListeners(listener)
+ .withJobDescriptorMergers(new TestJobDescriptionMerger())
+ .withVertx(externalVertx)
+ .build();
+
+ jobScheduler.init();
+ jobScheduler.schedule(new TestJobDescription(jobId,
ZonedDateTime.now().plus(Duration.ofSeconds(1))));
+ listener.waitForExecution();
+
+ assertThat(executor.getJobsExecuted()).hasSize(1);
+ assertThat(listener.isExecuted()).isTrue();
+
+ // close the scheduler — must NOT close the externally supplied
Vert.x
+ jobScheduler.close();
+
+ // external Vert.x is still usable (would throw/fail if already
closed)
+ assertThat(externalVertx.deploymentIDs()).isNotNull();
+ } finally {
+ externalVertx.close();
+ }
+ }
+
@Test
public void testBasicExactTimeFlow() throws Exception {
final String jobId = "1";
diff --git
a/kogito-quarkus/addons/persistence/jdbc/runtime/src/main/java/org/kie/kogito/persistence/quarkus/JDBCProcessInstancesFactory.java
b/kogito-quarkus/addons/persistence/jdbc/runtime/src/main/java/org/kie/kogito/persistence/quarkus/JDBCProcessInstancesFactory.java
index 34a485f15dc..1aed7784939 100644
---
a/kogito-quarkus/addons/persistence/jdbc/runtime/src/main/java/org/kie/kogito/persistence/quarkus/JDBCProcessInstancesFactory.java
+++
b/kogito-quarkus/addons/persistence/jdbc/runtime/src/main/java/org/kie/kogito/persistence/quarkus/JDBCProcessInstancesFactory.java
@@ -26,15 +26,21 @@ import javax.sql.DataSource;
import org.eclipse.microprofile.config.inject.ConfigProperty;
import org.kie.kogito.internal.process.runtime.HeadersPersistentConfig;
import org.kie.kogito.persistence.jdbc.AbstractProcessInstancesFactory;
+import org.kie.kogito.persistence.jdbc.JDBCProcessInstances;
+import org.kie.kogito.process.Process;
import org.kie.kogito.process.Processes;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.inject.Instance;
import jakarta.inject.Inject;
+import jakarta.transaction.TransactionSynchronizationRegistry;
@ApplicationScoped
public class JDBCProcessInstancesFactory extends
AbstractProcessInstancesFactory {
+ @Inject
+ Instance<TransactionSynchronizationRegistry> txSyncRegistry;
+
@Inject
public JDBCProcessInstancesFactory(DataSource dataSource,
@ConfigProperty(name = "kogito.persistence.optimistic.lock",
defaultValue = "false") Boolean lock,
@@ -48,4 +54,29 @@ public class JDBCProcessInstancesFactory extends
AbstractProcessInstancesFactory
public JDBCProcessInstancesFactory() {
}
+ @Override
+ public JDBCProcessInstances<?> createProcessInstances(Process<?> process) {
+ JDBCProcessInstances<?> instances =
super.createProcessInstances(process);
+ if (txSyncRegistry != null && txSyncRegistry.isResolvable()) {
+ TransactionSynchronizationRegistry registry = txSyncRegistry.get();
+ instances.setTransactionRegistrar(unlockAction -> {
+ if (registry.getTransactionStatus() !=
jakarta.transaction.Status.STATUS_NO_TRANSACTION) {
+ registry.registerInterposedSynchronization(new
jakarta.transaction.Synchronization() {
+ @Override
+ public void beforeCompletion() {
+ }
+
+ @Override
+ public void afterCompletion(int status) {
+ unlockAction.run();
+ }
+ });
+ } else {
+ unlockAction.run();
+ }
+ });
+ }
+ return instances;
+ }
+
}
diff --git
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/BranchExecutionCounter.java
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/BranchExecutionCounter.java
new file mode 100644
index 00000000000..a2a07d295c6
--- /dev/null
+++
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/BranchExecutionCounter.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.kie.kogito;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.inject.Singleton;
+
+/**
+ * Non-transactional attempt counter for the asyncsplit integration test.
+ * Incremented by BranchCounterWorkItemHandler on every execution attempt,
+ * including those whose surrounding JTA transaction is later rolled back.
+ * Because this state lives in a @Singleton CDI bean (not a DB row), it is
+ * never rolled back and therefore reliably reflects the total number of
+ * branch executions, not just committed ones.
+ */
+@Singleton
+public class BranchExecutionCounter {
+
+ private final AtomicInteger count = new AtomicInteger(0);
+
+ public int increment() {
+ return count.incrementAndGet();
+ }
+
+ public int get() {
+ return count.get();
+ }
+
+ public void reset() {
+ count.set(0);
+ }
+}
diff --git
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/wih/WIHRegister.java
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/BranchExecutionCounterResource.java
similarity index 64%
copy from
kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/wih/WIHRegister.java
copy to
kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/BranchExecutionCounterResource.java
index d0162926801..eb775c031a4 100644
---
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/wih/WIHRegister.java
+++
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/BranchExecutionCounterResource.java
@@ -16,15 +16,31 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.kie.kogito.wih;
-
-import org.kie.kogito.process.impl.DefaultWorkItemHandlerConfig;
+package org.kie.kogito;
import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.DELETE;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.MediaType;
@ApplicationScoped
-public class WIHRegister extends DefaultWorkItemHandlerConfig {
- {
- register("CustomTask", new CustomWorkItemHandler());
+@Path("/branch-counter")
+public class BranchExecutionCounterResource {
+
+ @Inject
+ BranchExecutionCounter counter;
+
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ public int get() {
+ return counter.get();
+ }
+
+ @DELETE
+ public void reset() {
+ counter.reset();
}
}
diff --git
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/wih/BranchCounterWorkItemHandler.java
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/wih/BranchCounterWorkItemHandler.java
new file mode 100644
index 00000000000..8689919fd62
--- /dev/null
+++
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/wih/BranchCounterWorkItemHandler.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.kie.kogito.wih;
+
+import java.util.Collections;
+import java.util.Optional;
+
+import org.kie.kogito.BranchExecutionCounter;
+import org.kie.kogito.internal.process.workitem.KogitoWorkItem;
+import org.kie.kogito.internal.process.workitem.KogitoWorkItemHandler;
+import org.kie.kogito.internal.process.workitem.KogitoWorkItemManager;
+import org.kie.kogito.internal.process.workitem.WorkItemTransition;
+import org.kie.kogito.process.workitems.impl.DefaultKogitoWorkItemHandler;
+
+import io.quarkus.arc.Arc;
+
+/**
+ * Work item handler for the asyncsplit race-condition reproducer.
+ * Sleeps for {@code waitTime} ms (to widen the race window) then increments
+ * a non-transactional {@link BranchExecutionCounter} singleton.
+ * Because the counter is outside the JTA transaction, its value reflects
+ * total execution attempts — not just committed ones — making it the correct
+ * observable for detecting optimistic-locking retries.
+ */
+public class BranchCounterWorkItemHandler extends DefaultKogitoWorkItemHandler
{
+
+ @Override
+ public Optional<WorkItemTransition>
activateWorkItemHandler(KogitoWorkItemManager manager, KogitoWorkItemHandler
handler,
+ KogitoWorkItem workItem, WorkItemTransition transition) {
+ Long waitTime = (Long) workItem.getParameter("waitTime");
+ if (waitTime != null && waitTime > 0) {
+ try {
+ Thread.sleep(waitTime);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ Arc.container().select(BranchExecutionCounter.class).get().increment();
+ return
Optional.of(handler.completeTransition(workItem.getPhaseStatus(),
Collections.emptyMap()));
+ }
+
+ @Override
+ public Optional<WorkItemTransition>
abortWorkItemHandler(KogitoWorkItemManager manager, KogitoWorkItemHandler
handler,
+ KogitoWorkItem workItem, WorkItemTransition transition) {
+ return Optional.of(handler.abortTransition(workItem.getPhaseStatus()));
+ }
+}
diff --git
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/wih/WIHRegister.java
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/wih/WIHRegister.java
index d0162926801..3e21a9902d9 100644
---
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/wih/WIHRegister.java
+++
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/java/org/kie/kogito/wih/WIHRegister.java
@@ -26,5 +26,6 @@ import jakarta.enterprise.context.ApplicationScoped;
public class WIHRegister extends DefaultWorkItemHandlerConfig {
{
register("CustomTask", new CustomWorkItemHandler());
+ register("BranchCounter", new BranchCounterWorkItemHandler());
}
}
diff --git
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/resources/asyncsplit.bpmn
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/resources/asyncsplit.bpmn
new file mode 100644
index 00000000000..281ed5e5351
--- /dev/null
+++
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/main/resources/asyncsplit.bpmn
@@ -0,0 +1,223 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+<!--
+ Reproducer for the parallel-async-split stale-lock bug:
+
+ Start → async TimerReplacement → parallel split → async Branch1 + async
Branch2
+ (both sleep waitTime ms and increment a non-transactional counter) →
parallel join → End
+
+ With waitTime=100 both branches finish nearly simultaneously and race to
signal the joining
+ gateway. The non-transactional BranchExecutionCounter reflects every
execution attempt,
+ including those whose JTA transaction was later rolled back. The test
asserts counter == 2,
+ meaning each branch ran exactly once with no retries.
+-->
+<bpmn2:definitions xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xmlns="http://www.omg.org/bpmn20";
+ xmlns:bpmn2="http://www.omg.org/spec/BPMN/20100524/MODEL";
+ xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI";
+ xmlns:bpsim="http://www.bpsim.org/schemas/1.0";
+ xmlns:dc="http://www.omg.org/spec/DD/20100524/DC";
+ xmlns:di="http://www.omg.org/spec/DD/20100524/DI";
+ xmlns:drools="http://www.jboss.org/drools";
+ id="_asyncsplit"
+ xsi:schemaLocation="http://www.omg.org/spec/BPMN/20100524/MODEL BPMN20.xsd
http://www.jboss.org/drools drools.xsd http://www.bpsim.org/schemas/1.0
bpsim.xsd http://www.omg.org/spec/DD/20100524/DC DC.xsd
http://www.omg.org/spec/DD/20100524/DI DI.xsd"
+ exporter="jBPM Process Modeler" exporterVersion="2.0"
targetNamespace="http://www.omg.org/bpmn20";>
+
+ <bpmn2:itemDefinition id="_waitTimeItem" structureRef="Long" />
+
+ <bpmn2:collaboration id="_collaboration" name="Default Collaboration">
+ <bpmn2:participant id="_participant" name="Pool Participant"
processRef="asyncsplit"/>
+ </bpmn2:collaboration>
+
+ <bpmn2:process id="asyncsplit" drools:packageName="org.kie.kogito"
drools:version="1.0" drools:adHoc="false"
+ name="asyncsplit" isExecutable="true" processType="Public">
+
+ <bpmn2:property id="waitTime" itemSubjectRef="_waitTimeItem"
name="waitTime" />
+
+ <bpmn2:startEvent id="_Start">
+ <bpmn2:outgoing>_f1</bpmn2:outgoing>
+ </bpmn2:startEvent>
+
+ <!-- Async script that initialises nothing — kept async to reproduce the
job-scheduling pattern -->
+ <bpmn2:scriptTask id="_Timer" name="TimerReplacement"
scriptFormat="http://www.java.com/java";>
+ <bpmn2:extensionElements>
+ <drools:metaData
name="elementname"><drools:metaValue><![CDATA[TimerReplacement]]></drools:metaValue></drools:metaData>
+ <drools:metaData
name="customAsync"><drools:metaValue><![CDATA[true]]></drools:metaValue></drools:metaData>
+ </bpmn2:extensionElements>
+ <bpmn2:incoming>_f1</bpmn2:incoming>
+ <bpmn2:outgoing>_f2</bpmn2:outgoing>
+ <bpmn2:script><![CDATA[// intentionally empty]]></bpmn2:script>
+ </bpmn2:scriptTask>
+
+ <bpmn2:parallelGateway id="_Split" gatewayDirection="Diverging">
+ <bpmn2:incoming>_f2</bpmn2:incoming>
+ <bpmn2:outgoing>_f3</bpmn2:outgoing>
+ <bpmn2:outgoing>_f4</bpmn2:outgoing>
+ </bpmn2:parallelGateway>
+
+ <!--
+ Branch1: async work item that sleeps waitTime ms then increments the
non-transactional counter.
+ customAsync=true causes this to be scheduled as a separate job, creating
the race condition.
+ -->
+ <bpmn2:task id="_Branch1" drools:taskName="BranchCounter" name="Branch1">
+ <bpmn2:extensionElements>
+ <drools:metaData
name="elementname"><drools:metaValue><![CDATA[Branch1]]></drools:metaValue></drools:metaData>
+ <drools:metaData
name="customAsync"><drools:metaValue><![CDATA[true]]></drools:metaValue></drools:metaData>
+ </bpmn2:extensionElements>
+ <bpmn2:incoming>_f3</bpmn2:incoming>
+ <bpmn2:outgoing>_f5a</bpmn2:outgoing>
+ <bpmn2:ioSpecification>
+ <bpmn2:dataInput id="_Branch1_waitTimeInput" drools:dtype="Long"
name="waitTime"/>
+ <bpmn2:dataInput id="_Branch1_TaskNameInput" drools:dtype="Object"
name="TaskName"/>
+ <bpmn2:inputSet>
+ <bpmn2:dataInputRefs>_Branch1_waitTimeInput</bpmn2:dataInputRefs>
+ <bpmn2:dataInputRefs>_Branch1_TaskNameInput</bpmn2:dataInputRefs>
+ </bpmn2:inputSet>
+ <bpmn2:outputSet/>
+ </bpmn2:ioSpecification>
+ <bpmn2:dataInputAssociation>
+ <bpmn2:sourceRef>waitTime</bpmn2:sourceRef>
+ <bpmn2:targetRef>_Branch1_waitTimeInput</bpmn2:targetRef>
+ </bpmn2:dataInputAssociation>
+ <bpmn2:dataInputAssociation>
+ <bpmn2:targetRef>_Branch1_TaskNameInput</bpmn2:targetRef>
+ <bpmn2:assignment>
+ <bpmn2:from
xsi:type="bpmn2:tFormalExpression"><![CDATA[BranchCounter]]></bpmn2:from>
+ <bpmn2:to
xsi:type="bpmn2:tFormalExpression"><![CDATA[_Branch1_TaskNameInput]]></bpmn2:to>
+ </bpmn2:assignment>
+ </bpmn2:dataInputAssociation>
+ </bpmn2:task>
+
+ <!-- Branch2: identical to Branch1, races it to the parallel join -->
+ <bpmn2:task id="_Branch2" drools:taskName="BranchCounter" name="Branch2">
+ <bpmn2:extensionElements>
+ <drools:metaData
name="elementname"><drools:metaValue><![CDATA[Branch2]]></drools:metaValue></drools:metaData>
+ <drools:metaData
name="customAsync"><drools:metaValue><![CDATA[true]]></drools:metaValue></drools:metaData>
+ </bpmn2:extensionElements>
+ <bpmn2:incoming>_f4</bpmn2:incoming>
+ <bpmn2:outgoing>_f5b</bpmn2:outgoing>
+ <bpmn2:ioSpecification>
+ <bpmn2:dataInput id="_Branch2_waitTimeInput" drools:dtype="Long"
name="waitTime"/>
+ <bpmn2:dataInput id="_Branch2_TaskNameInput" drools:dtype="Object"
name="TaskName"/>
+ <bpmn2:inputSet>
+ <bpmn2:dataInputRefs>_Branch2_waitTimeInput</bpmn2:dataInputRefs>
+ <bpmn2:dataInputRefs>_Branch2_TaskNameInput</bpmn2:dataInputRefs>
+ </bpmn2:inputSet>
+ <bpmn2:outputSet/>
+ </bpmn2:ioSpecification>
+ <bpmn2:dataInputAssociation>
+ <bpmn2:sourceRef>waitTime</bpmn2:sourceRef>
+ <bpmn2:targetRef>_Branch2_waitTimeInput</bpmn2:targetRef>
+ </bpmn2:dataInputAssociation>
+ <bpmn2:dataInputAssociation>
+ <bpmn2:targetRef>_Branch2_TaskNameInput</bpmn2:targetRef>
+ <bpmn2:assignment>
+ <bpmn2:from
xsi:type="bpmn2:tFormalExpression"><![CDATA[BranchCounter]]></bpmn2:from>
+ <bpmn2:to
xsi:type="bpmn2:tFormalExpression"><![CDATA[_Branch2_TaskNameInput]]></bpmn2:to>
+ </bpmn2:assignment>
+ </bpmn2:dataInputAssociation>
+ </bpmn2:task>
+
+ <bpmn2:parallelGateway id="_Join" gatewayDirection="Converging">
+ <bpmn2:incoming>_f5a</bpmn2:incoming>
+ <bpmn2:incoming>_f5b</bpmn2:incoming>
+ <bpmn2:outgoing>_f6</bpmn2:outgoing>
+ </bpmn2:parallelGateway>
+
+ <bpmn2:endEvent id="_End">
+ <bpmn2:incoming>_f6</bpmn2:incoming>
+ </bpmn2:endEvent>
+
+ <bpmn2:sequenceFlow id="_f1" sourceRef="_Start" targetRef="_Timer" />
+ <bpmn2:sequenceFlow id="_f2" sourceRef="_Timer" targetRef="_Split" />
+ <bpmn2:sequenceFlow id="_f3" sourceRef="_Split" targetRef="_Branch1" />
+ <bpmn2:sequenceFlow id="_f4" sourceRef="_Split" targetRef="_Branch2" />
+ <bpmn2:sequenceFlow id="_f5a" sourceRef="_Branch1" targetRef="_Join" />
+ <bpmn2:sequenceFlow id="_f5b" sourceRef="_Branch2" targetRef="_Join" />
+ <bpmn2:sequenceFlow id="_f6" sourceRef="_Join" targetRef="_End" />
+ </bpmn2:process>
+ <bpmndi:BPMNDiagram>
+ <bpmndi:BPMNPlane bpmnElement="asyncsplit">
+ <bpmndi:BPMNShape id="shape__Start" bpmnElement="_Start">
+ <dc:Bounds height="56" width="56" x="100" y="140"/>
+ </bpmndi:BPMNShape>
+ <bpmndi:BPMNShape id="shape__Timer" bpmnElement="_Timer">
+ <dc:Bounds height="102" width="154" x="200" y="117"/>
+ </bpmndi:BPMNShape>
+ <bpmndi:BPMNShape id="shape__Split" bpmnElement="_Split">
+ <dc:Bounds height="56" width="56" x="400" y="140"/>
+ </bpmndi:BPMNShape>
+ <bpmndi:BPMNShape id="shape__Branch1" bpmnElement="_Branch1">
+ <dc:Bounds height="102" width="154" x="500" y="70"/>
+ </bpmndi:BPMNShape>
+ <bpmndi:BPMNShape id="shape__Branch2" bpmnElement="_Branch2">
+ <dc:Bounds height="102" width="154" x="500" y="210"/>
+ </bpmndi:BPMNShape>
+ <bpmndi:BPMNShape id="shape__Join" bpmnElement="_Join">
+ <dc:Bounds height="56" width="56" x="700" y="140"/>
+ </bpmndi:BPMNShape>
+ <bpmndi:BPMNShape id="shape__End" bpmnElement="_End">
+ <dc:Bounds height="56" width="56" x="800" y="140"/>
+ </bpmndi:BPMNShape>
+ <bpmndi:BPMNEdge id="edge__f1" bpmnElement="_f1">
+ <di:waypoint x="156" y="168"/>
+ <di:waypoint x="200" y="168"/>
+ </bpmndi:BPMNEdge>
+ <bpmndi:BPMNEdge id="edge__f2" bpmnElement="_f2">
+ <di:waypoint x="354" y="168"/>
+ <di:waypoint x="400" y="168"/>
+ </bpmndi:BPMNEdge>
+ <bpmndi:BPMNEdge id="edge__f3" bpmnElement="_f3">
+ <di:waypoint x="428" y="140"/>
+ <di:waypoint x="428" y="121"/>
+ <di:waypoint x="500" y="121"/>
+ </bpmndi:BPMNEdge>
+ <bpmndi:BPMNEdge id="edge__f4" bpmnElement="_f4">
+ <di:waypoint x="428" y="196"/>
+ <di:waypoint x="428" y="261"/>
+ <di:waypoint x="500" y="261"/>
+ </bpmndi:BPMNEdge>
+ <bpmndi:BPMNEdge id="edge__f5a" bpmnElement="_f5a">
+ <di:waypoint x="654" y="121"/>
+ <di:waypoint x="728" y="121"/>
+ <di:waypoint x="728" y="140"/>
+ </bpmndi:BPMNEdge>
+ <bpmndi:BPMNEdge id="edge__f5b" bpmnElement="_f5b">
+ <di:waypoint x="654" y="261"/>
+ <di:waypoint x="728" y="261"/>
+ <di:waypoint x="728" y="196"/>
+ </bpmndi:BPMNEdge>
+ <bpmndi:BPMNEdge id="edge__f6" bpmnElement="_f6">
+ <di:waypoint x="756" y="168"/>
+ <di:waypoint x="800" y="168"/>
+ </bpmndi:BPMNEdge>
+ </bpmndi:BPMNPlane>
+ </bpmndi:BPMNDiagram>
+ <bpmn2:relationship type="BPSimData">
+ <bpmn2:extensionElements>
+ <bpsim:BPSimData>
+ <bpsim:Scenario id="default" name="Simulationscenario">
+ <bpsim:ScenarioParameters/>
+ </bpsim:Scenario>
+ </bpsim:BPSimData>
+ </bpmn2:extensionElements>
+ <bpmn2:source>_asyncsplit</bpmn2:source>
+ <bpmn2:target>_asyncsplit</bpmn2:target>
+ </bpmn2:relationship>
+</bpmn2:definitions>
diff --git
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/test/java/org/kie/kogito/it/OptimisticLockingTest.java
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/test/java/org/kie/kogito/it/OptimisticLockingTest.java
index 04028e27898..ef411737b05 100644
---
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/test/java/org/kie/kogito/it/OptimisticLockingTest.java
+++
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-processes-persistence-common/src/test/java/org/kie/kogito/it/OptimisticLockingTest.java
@@ -18,6 +18,8 @@
*/
package org.kie.kogito.it;
+import java.util.Map;
+
import org.junit.jupiter.api.Test;
import io.quarkus.test.junit.TestProfile;
@@ -27,6 +29,7 @@ import static io.restassured.RestAssured.given;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.Matchers.emptyOrNullString;
+import static org.hamcrest.Matchers.equalTo;
@TestProfile(OptimisticLockingProfile.class)
public abstract class OptimisticLockingTest extends PersistenceTest {
@@ -50,4 +53,55 @@ public abstract class OptimisticLockingTest extends
PersistenceTest {
.then()
.statusCode(404));
}
+
+ /**
+ * Verifies that a process with an async parallel split completes its
parallel branches
+ * exactly once each, without a ProcessInstanceOptimisticLockingException
caused by a
+ * stale optimistic-lock version read under concurrent lock access.
+ * <p>
+ * The asyncsplit process: async TimerReplacement → parallel split → async
Branch1 +
+ * async Branch2 (both sleep {@code waitTime} ms and increment a
non-transactional counter)
+ * → parallel join → end.
+ * <p>
+ * With {@code waitTime=100} both worker threads complete nearly
simultaneously, triggering
+ * the race. Before the fix, the second thread would read a stale
optimistic-lock version
+ * and fail with a locking exception, causing a retry that increments the
counter a third
+ * time. After the fix the lock is deferred until after the transaction
commits, so each
+ * branch runs exactly once and the counter equals 2.
+ * <p>
+ * The counter is non-transactional (an AtomicInteger in a @Singleton CDI
bean) so it
+ * reflects every execution attempt, including those rolled back by a
failed transaction.
+ */
+ @Test
+ void testAsyncSplitWithParallelAsyncTasks() {
+ // Reset the non-transactional counter before the test run.
+ given().when().delete("/branch-counter").then().statusCode(204);
+
+ final String pid = given().contentType(ContentType.JSON)
+ .when()
+ .body(Map.of("waitTime", 100L))
+ .post("/asyncsplit")
+ .then()
+ .statusCode(201)
+ .header("Location", not(emptyOrNullString()))
+ .body("id", not(emptyOrNullString()))
+ .extract()
+ .path("id");
+
+ // Wait for the process to complete (it ends automatically after the
parallel join).
+ await().atMost(TIMEOUT)
+ .untilAsserted(() -> given().contentType(ContentType.JSON)
+ .when()
+ .get("/asyncsplit/{pid}", pid)
+ .then()
+ .statusCode(404));
+
+ // Counter must be exactly 2: Branch1 ran once and Branch2 ran once.
+ // A value > 2 means a retry occurred due to the stale-lock bug.
+ given().when()
+ .get("/branch-counter")
+ .then()
+ .statusCode(200)
+ .body(equalTo("2"));
+ }
}
diff --git
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-quarkus-processes-jdbc/pom.xml
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-quarkus-processes-jdbc/pom.xml
index e09843804df..327a56c0acc 100644
---
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-quarkus-processes-jdbc/pom.xml
+++
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/integration-tests-quarkus-processes-jdbc/pom.xml
@@ -51,6 +51,14 @@
<groupId>org.kie</groupId>
<artifactId>kie-addons-quarkus-persistence-jdbc</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.kie</groupId>
+ <artifactId>kogito-addons-quarkus-embedded-jobs</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.kie</groupId>
+ <artifactId>kogito-addons-quarkus-embedded-jobs-jpa</artifactId>
+ </dependency>
<dependency>
<groupId>org.kie</groupId>
<artifactId>kie-addons-quarkus-process-management</artifactId>
diff --git
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/pom.xml
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/pom.xml
index 51ffc80caad..7914f5f431c 100644
---
a/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/pom.xml
+++
b/kogito-quarkus/integration-tests/integration-tests-quarkus-processes-persistence/pom.xml
@@ -29,6 +29,23 @@
<artifactId>integration-tests-quarkus-processes-persistence</artifactId>
<name>Kogito :: Integration Tests :: Quarkus :: Processes ::
Persistence</name>
+ <properties>
+
<allowedPomsList>org.kie.kogito:kogito-quarkus-bom,org.kie.kogito:integration-tests-quarkus-processes-persistence</allowedPomsList>
+ <allowed_GA>org.kie.kogito:kogito-apps-quarkus-bom</allowed_GA>
+ </properties>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.kie.kogito</groupId>
+ <artifactId>kogito-apps-quarkus-bom</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
<modules>
<module>integration-tests-processes-persistence-common</module>
<module>integration-tests-quarkus-processes-postgresql</module>
diff --git
a/kogito-springboot/addons/persistence/jdbc/src/main/java/org/kie/kogito/persistence/springboot/JDBCProcessInstancesFactory.java
b/kogito-springboot/addons/persistence/jdbc/src/main/java/org/kie/kogito/persistence/springboot/JDBCProcessInstancesFactory.java
index 0cf18a58d2e..0d9119d33cf 100644
---
a/kogito-springboot/addons/persistence/jdbc/src/main/java/org/kie/kogito/persistence/springboot/JDBCProcessInstancesFactory.java
+++
b/kogito-springboot/addons/persistence/jdbc/src/main/java/org/kie/kogito/persistence/springboot/JDBCProcessInstancesFactory.java
@@ -24,12 +24,16 @@ import javax.sql.DataSource;
import org.kie.kogito.internal.process.runtime.HeadersPersistentConfig;
import org.kie.kogito.persistence.jdbc.AbstractProcessInstancesFactory;
+import org.kie.kogito.persistence.jdbc.JDBCProcessInstances;
+import org.kie.kogito.process.Process;
import org.kie.kogito.process.Processes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.datasource.TransactionAwareDataSourceProxy;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;
+import org.springframework.transaction.support.TransactionSynchronization;
+import
org.springframework.transaction.support.TransactionSynchronizationManager;
@Component
public class JDBCProcessInstancesFactory extends
AbstractProcessInstancesFactory {
@@ -49,4 +53,22 @@ public class JDBCProcessInstancesFactory extends
AbstractProcessInstancesFactory
super(new TransactionAwareDataSourceProxy(dataSource), lock, new
HeadersPersistentConfig(headersEnabled, headersExcluded), dataIsolationEnabled
? processes : null);
}
+ @Override
+ public JDBCProcessInstances<?> createProcessInstances(Process<?> process) {
+ JDBCProcessInstances<?> instances =
super.createProcessInstances(process);
+ instances.setTransactionRegistrar(unlockAction -> {
+ if (TransactionSynchronizationManager.isSynchronizationActive()) {
+ TransactionSynchronizationManager.registerSynchronization(new
TransactionSynchronization() {
+ @Override
+ public void afterCompletion(int status) {
+ unlockAction.run();
+ }
+ });
+ } else {
+ unlockAction.run();
+ }
+ });
+ return instances;
+ }
+
}
diff --git
a/kogito-springboot/starters/kogito-processes-spring-boot-starter/src/main/java/org/kie/kogito/process/KogitoBeanProducer.java
b/kogito-springboot/starters/kogito-processes-spring-boot-starter/src/main/java/org/kie/kogito/process/KogitoBeanProducer.java
index 7814bfe5236..35c4bbaf1b1 100644
---
a/kogito-springboot/starters/kogito-processes-spring-boot-starter/src/main/java/org/kie/kogito/process/KogitoBeanProducer.java
+++
b/kogito-springboot/starters/kogito-processes-spring-boot-starter/src/main/java/org/kie/kogito/process/KogitoBeanProducer.java
@@ -90,7 +90,13 @@ public class KogitoBeanProducer {
@Bean
@ConditionalOnMissingBean(Vertx.class)
- Vertx vertx() {
+ Vertx vertxMutiny() {
return Vertx.vertx();
}
+
+ @Bean
+ @ConditionalOnMissingBean(io.vertx.core.Vertx.class)
+ io.vertx.core.Vertx vertx() {
+ return io.vertx.core.Vertx.vertx();
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
