This is an automated email from the ASF dual-hosted git repository.
pefernan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-kie-kogito-apps.git
The following commit(s) were added to refs/heads/main by this push:
new a4b8aa613 [incubator-kie-issues#2311] Handle reconciliation of RETRY
or ERROR jobs in a new transaction (#2333)
a4b8aa613 is described below
commit a4b8aa613b6fb9698b022e08d0f0f9537346e512
Author: Martin Weiler <[email protected]>
AuthorDate: Wed Jul 1 01:21:20 2026 -0600
[incubator-kie-issues#2311] Handle reconciliation of RETRY or ERROR jobs in
a new transaction (#2333)
* [incubator-kie-issues#2311] Handle reconciliation of RETRY or ERROR jobs
in a new transaction
* Mark current transaction for rollback
* Replace reflection code with platform specific implementations
* Add test cases to verify transactional behavior with rollbacks
* Ensure jobs are executed in parallel threads (required for Spring Boot)
* Fix test failures
* Review feedback
* Review feedback, simplify logic around rescheduling
* Fix usage of deprecated method
* Simplification
---
.../kogito/app/jobs/api/JobSchedulerBuilder.java | 3 +
.../kogito/app/jobs/impl/VertxJobScheduler.java | 90 ++++++++++--
.../ErrorHandlingJobTimeoutInterceptor.java | 4 +-
.../jobs/spi/NoOpTransactionRollbackMarker.java | 38 ++---
.../app/jobs/spi/TransactionRollbackMarker.java | 40 ++++++
.../app/jobs/quarkus/QuarkusJobsService.java | 5 +
.../quarkus/QuarkusTransactionRollbackMarker.java | 67 +++++++++
.../app/jobs/jpa/quarkus/MockDataRepository.java | 115 +++++++++++++++
.../quarkus/QuarkusTransactionRollbackTest.java | 157 +++++++++++++++++++++
.../app/jobs/jpa/quarkus/TestJobExecutor.java | 10 ++
.../SpringBootTransactionRollbackMarker.java | 61 ++++++++
.../app/jobs/springboot/SpringbootJobsService.java | 9 +-
.../TransactionJobTimeoutInterceptor.java | 17 +++
.../app/jobs/springboot/MockDataRepository.java | 98 +++++++++++++
.../SpringBootTransactionRollbackTest.java | 122 ++++++++++++++++
.../app/jobs/springboot/TestJobExecutor.java | 10 ++
16 files changed, 808 insertions(+), 38 deletions(-)
diff --git
a/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/api/JobSchedulerBuilder.java
b/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/api/JobSchedulerBuilder.java
index 6380da172..8376af625 100644
---
a/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/api/JobSchedulerBuilder.java
+++
b/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/api/JobSchedulerBuilder.java
@@ -22,6 +22,7 @@ import org.kie.kogito.app.jobs.impl.VertxJobScheduler;
import org.kie.kogito.app.jobs.integrations.JobExceptionDetailsExtractor;
import org.kie.kogito.app.jobs.spi.JobContextFactory;
import org.kie.kogito.app.jobs.spi.JobStore;
+import org.kie.kogito.app.jobs.spi.TransactionRollbackMarker;
import org.kie.kogito.event.EventPublisher;
public interface JobSchedulerBuilder {
@@ -61,4 +62,6 @@ public interface JobSchedulerBuilder {
JobSchedulerBuilder withJobDescriptorMergers(JobDescriptionMerger...
jobDescriptionMergers);
JobSchedulerBuilder
withExceptionDetailsExtractor(JobExceptionDetailsExtractor
exceptionDetailsExtractor);
+
+ JobSchedulerBuilder
withTransactionRollbackMarker(TransactionRollbackMarker
transactionRollbackMarker);
}
diff --git
a/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java
b/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java
index ebc38074a..d9ecceaff 100644
---
a/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java
+++
b/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java
@@ -47,6 +47,8 @@ import
org.kie.kogito.app.jobs.integrations.UserTaskInstanceJobDescriptorMerger;
import org.kie.kogito.app.jobs.spi.JobContext;
import org.kie.kogito.app.jobs.spi.JobContextFactory;
import org.kie.kogito.app.jobs.spi.JobStore;
+import org.kie.kogito.app.jobs.spi.NoOpTransactionRollbackMarker;
+import org.kie.kogito.app.jobs.spi.TransactionRollbackMarker;
import org.kie.kogito.app.jobs.spi.memory.MemoryJobContextFactory;
import org.kie.kogito.app.jobs.spi.memory.MemoryJobStore;
import org.kie.kogito.event.DataEvent;
@@ -115,6 +117,8 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
private JobExceptionDetailsExtractor exceptionDetailsExtractor;
+ private TransactionRollbackMarker transactionRollbackMarker;
+
public class VertxJobSchedulerBuilder implements JobSchedulerBuilder {
@Override
@@ -213,6 +217,12 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
return this;
}
+ @Override
+ public JobSchedulerBuilder
withTransactionRollbackMarker(TransactionRollbackMarker
transactionRollbackMarker) {
+ VertxJobScheduler.this.transactionRollbackMarker =
transactionRollbackMarker;
+ return this;
+ }
+
}
public VertxJobScheduler() {
@@ -242,6 +252,7 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
this.refreshJobsInterval = 1000L;
this.retryInterval = 10 * 1000L; // ten seconds
this.maxRefreshJobsIntervalWindow = 5 * 60 * 1000L; // every 5 minute
+ this.transactionRollbackMarker = new NoOpTransactionRollbackMarker();
}
@Override
@@ -390,7 +401,7 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
private void timeout(Long timerId, String jobId) {
LOG.debug("Executing timeout with timer Id {} and jobId {}", timerId,
jobId);
- workerExecutor.executeBlocking(newTimeoutTask(timerId, jobId));
+ workerExecutor.executeBlocking(newTimeoutTask(timerId, jobId), false);
}
private Callable<JobTimeoutExecution> newTimeoutTask(Long timerId, String
jobId) {
@@ -421,6 +432,10 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
return new JobTimeoutExecution(nextJobDetails);
} catch (Exception exception) {
LOG.trace("Timeout {} with jobId {} will be retried if
possible", timerId, jobId, exception);
+
+ // Mark the current transaction for rollback
+ markTransactionForRollbackWhenEnabled();
+
JobDetails nextJobDetails = doRetryOrError(jobDetails,
exception);
reconcileScheduling(timerId, jobContext, nextJobDetails);
jobSchedulerListeners.forEach(l ->
l.onFailure(jobDetails));
@@ -435,6 +450,55 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
return current;
}
+ /**
+ * Handles retry scheduling for RETRY and ERROR job statuses.
+ * If there's an active transaction that was marked for rollback,
schedules the retry
+ * in a new transaction. Otherwise, persists the job immediately.
+ */
+ private void scheduleRetry(JobContext jobContext, JobDetails jobDetails) {
+ String jobId = jobDetails.getId();
+ JobStatus status = jobDetails.getStatus();
+
+ // Check if we need to execute in a new transaction
+ if (transactionRollbackMarker != null &&
transactionRollbackMarker.isTransactionActive()) {
+ // Current transaction will be rolled back, so schedule retry in a
new transaction
+ LOG.trace("Job {} with status {} will be scheduled in new
transaction", jobId, status);
+
+ JobContext newJobContext = jobContextFactory.newContext();
+
+ Callable<JobTimeoutExecution> newTransactionTask = () ->
persistAndSchedule(newJobContext, jobDetails, status);
+
+ // Apply transaction interceptors to ensure this runs in a new
transaction
+ for (JobTimeoutInterceptor interceptor : interceptors) {
+ newTransactionTask =
interceptor.chainIntercept(newTransactionTask);
+ }
+
+ vertx.executeBlocking(newTransactionTask, false)
+ .onFailure(e -> LOG.error("Async retry scheduling failed
for job {}", jobId, e));
+ } else {
+ // No active transaction or transaction not marked for rollback
+ LOG.trace("Job {} with status {} will be scheduled", jobId,
status);
+
+ try {
+ persistAndSchedule(jobContext, jobDetails, status);
+ } catch (Exception e) {
+ LOG.error("Failed to schedule retry for job {}", jobId, e);
+ }
+ }
+ }
+
+ private JobTimeoutExecution persistAndSchedule(JobContext jobContext,
JobDetails jobDetails, JobStatus status) throws Exception {
+ jobStore.update(jobContext, jobDetails);
+ fireEvents(jobDetails);
+ // Schedule timer after persistence
+ if (status == JobStatus.RETRY) {
+ addOrUpdateTxTimer(jobDetails);
+ } else {
+ removeTxTimer(jobDetails);
+ }
+ return new JobTimeoutExecution(jobDetails);
+ }
+
private void reconcileScheduling(Long timerId, JobContext jobContext,
JobDetails nextJobDetails) {
String jobId = nextJobDetails.getId();
switch (nextJobDetails.getStatus()) {
@@ -445,15 +509,13 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
jobStore.remove(jobContext, jobId);
break;
case SCHEDULED:
- case RETRY:
LOG.trace("Timeout {} with jobId {} will be updated and
scheduled", timerId, jobId);
addOrUpdateTxTimer(nextJobDetails);
jobStore.update(jobContext, nextJobDetails);
break;
+ case RETRY:
case ERROR:
- LOG.trace("Timeout {} with jobId {} will be set to error",
timerId, jobId);
- removeTxTimer(nextJobDetails);
- jobStore.update(jobContext, nextJobDetails);
+ scheduleRetry(jobContext, nextJobDetails);
break;
default:
LOG.trace("Timeout {} with jobId {} is RUNNING and should not
happen", timerId, jobId);
@@ -553,7 +615,7 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
private JobDetails doRetryOrError(JobDetails jobDetails, Exception
exception) {
Integer retryCounter = jobDetails.getRetries();
- // First, create the next JobDetails (RETRY or ERROR) without firing
events
+ // Create the next JobDetails (RETRY or ERROR)
JobDetails nextJobDetails;
if (retryCounter < this.maxNumberOfRetries) {
LOG.trace("doRetryOrError: Job {} will be retried (retry {} of
{})", jobDetails.getId(), retryCounter + 1, maxNumberOfRetries);
@@ -562,11 +624,10 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
LOG.trace("doRetryOrError: Job {} exceeded max retries ({}) and
will transition to ERROR", jobDetails.getId(), maxNumberOfRetries);
nextJobDetails = createErrorJobDetails(jobDetails);
}
- // Then extract and set exception details BEFORE firing events
+ // Extract and set exception details
extractAndSetExceptionDetails(nextJobDetails, exception);
LOG.trace("doRetryOrError: Created {} with exception details: {}",
nextJobDetails.getStatus(), nextJobDetails);
- // Finally fire events with exception details already set
- fireEvents(nextJobDetails);
+ // Events will be fired in scheduleRetry() after persistence
return nextJobDetails;
}
@@ -649,4 +710,15 @@ public class VertxJobScheduler implements JobScheduler,
Handler<Long> {
}
}
+ /**
+ * Marks the current transaction for rollback using the injected
TransactionRollbackMarker.
+ */
+ private void markTransactionForRollbackWhenEnabled() {
+ if (transactionRollbackMarker != null) {
+ transactionRollbackMarker.markForRollback();
+ } else {
+ LOG.warn("No TransactionRollbackMarker configured. Transaction
rollback marking is not supported.");
+ }
+ }
+
}
diff --git
a/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/integrations/ErrorHandlingJobTimeoutInterceptor.java
b/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/integrations/ErrorHandlingJobTimeoutInterceptor.java
index 8e1308bab..897c70e04 100644
---
a/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/integrations/ErrorHandlingJobTimeoutInterceptor.java
+++
b/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/integrations/ErrorHandlingJobTimeoutInterceptor.java
@@ -41,7 +41,9 @@ public class ErrorHandlingJobTimeoutInterceptor implements
JobTimeoutInterceptor
@Override
public Integer priority() {
- return 50;
+ // Priority 5 ensures this runs OUTSIDE the transaction interceptor
(priority 10)
+ // This allows error handling to occur after transaction rollback
+ return 5;
}
@Override
diff --git
a/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/TestJobExecutor.java
b/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/spi/NoOpTransactionRollbackMarker.java
similarity index 52%
copy from
kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/TestJobExecutor.java
copy to
jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/spi/NoOpTransactionRollbackMarker.java
index 8cec79896..297dc7d51 100644
---
a/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/TestJobExecutor.java
+++
b/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/spi/NoOpTransactionRollbackMarker.java
@@ -16,41 +16,27 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.kie.kogito.app.jobs.jpa.quarkus;
+package org.kie.kogito.app.jobs.spi;
-import org.kie.kogito.app.jobs.api.JobExecutor;
-import org.kie.kogito.jobs.service.model.JobDetails;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import jakarta.inject.Singleton;
-
-@Singleton
-public class TestJobExecutor implements JobExecutor {
+/**
+ * No-op implementation of TransactionRollbackMarker that does nothing.
+ */
+public class NoOpTransactionRollbackMarker implements
TransactionRollbackMarker {
- private Logger LOG = LoggerFactory.getLogger(TestJobExecutor.class);
- private int numberOfFailures;
+ private static final Logger LOG =
LoggerFactory.getLogger(NoOpTransactionRollbackMarker.class);
@Override
- public boolean accept(JobDetails jobDescription) {
- return true;
+ public void markForRollback() {
+ LOG.warn("No platform-specific TransactionRollbackMarker
implementation found. " +
+ "Transaction rollback marking is not supported. " +
+ "Please ensure a Quarkus or Spring Boot specific
implementation is available.");
}
@Override
- public void execute(JobDetails jobDescription) {
- LOG.info("executing {}", jobDescription);
- if (numberOfFailures > 0) {
- --numberOfFailures;
- throw new RuntimeException();
- }
- }
-
- public void setNumberOfFailures(int numberOfFailures) {
- this.numberOfFailures = numberOfFailures;
+ public boolean isTransactionActive() {
+ return false;
}
-
- public void reset() {
- numberOfFailures = 0;
- }
-
}
diff --git
a/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/spi/TransactionRollbackMarker.java
b/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/spi/TransactionRollbackMarker.java
new file mode 100644
index 000000000..5a10a5164
--- /dev/null
+++
b/jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/spi/TransactionRollbackMarker.java
@@ -0,0 +1,40 @@
+/*
+ * 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.app.jobs.spi;
+
+/**
+ * Interface for marking the current transaction for rollback.
+ * This is used when a job execution fails and needs to be retried, ensuring
that
+ * any changes made during the failed execution are rolled back before the
retry attempt.
+ */
+public interface TransactionRollbackMarker {
+
+ /**
+ * Marks the current transaction for rollback.
+ * If no transaction is active, this method should do nothing.
+ */
+ void markForRollback();
+
+ /**
+ * Checks if a transaction is currently active.
+ *
+ * @return true if a transaction is active, false otherwise
+ */
+ boolean isTransactionActive();
+}
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 77be00aaa..801380061 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
@@ -31,6 +31,7 @@ import
org.kie.kogito.app.jobs.integrations.UserTaskInstanceJobDescriptionJobIns
import org.kie.kogito.app.jobs.quarkus.resource.RestApiConstants;
import org.kie.kogito.app.jobs.spi.JobContextFactory;
import org.kie.kogito.app.jobs.spi.JobStore;
+import org.kie.kogito.app.jobs.spi.TransactionRollbackMarker;
import org.kie.kogito.event.EventPublisher;
import org.kie.kogito.handler.ExceptionHandler;
import org.kie.kogito.jobs.JobDescription;
@@ -94,6 +95,9 @@ public class QuarkusJobsService implements JobsService {
@Inject
WrappingConditionalJobExceptionDetailsExtractor exceptionDetailsExtractor;
+ @Inject
+ TransactionRollbackMarker transactionRollbackMarker;
+
@PostConstruct
public void init() {
this.jobScheduler = JobSchedulerBuilder.newJobSchedulerBuilder()
@@ -114,6 +118,7 @@ public class QuarkusJobsService implements JobsService {
new TransactionJobTimeoutInterceptor(),
new
ErrorHandlingJobTimeoutInterceptor(exceptionHandlers.stream().toList()))
.withExceptionDetailsExtractor(exceptionDetailsExtractor)
+ .withTransactionRollbackMarker(transactionRollbackMarker)
.withNumberOfWorkerThreads(numberOfWorkerThreads)
.withJobSynchronization(new JobSynchronization() {
diff --git
a/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/quarkus/QuarkusTransactionRollbackMarker.java
b/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/quarkus/QuarkusTransactionRollbackMarker.java
new file mode 100644
index 000000000..ab03bdeee
--- /dev/null
+++
b/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/quarkus/QuarkusTransactionRollbackMarker.java
@@ -0,0 +1,67 @@
+/*
+ * 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.app.jobs.quarkus;
+
+import org.kie.kogito.app.jobs.spi.TransactionRollbackMarker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.transaction.Status;
+import jakarta.transaction.SystemException;
+import jakarta.transaction.TransactionManager;
+
+@ApplicationScoped
+public class QuarkusTransactionRollbackMarker implements
TransactionRollbackMarker {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(QuarkusTransactionRollbackMarker.class);
+
+ private final TransactionManager transactionManager;
+
+ public QuarkusTransactionRollbackMarker(TransactionManager
transactionManager) {
+ this.transactionManager = transactionManager;
+ }
+
+ @Override
+ public void markForRollback() {
+ try {
+ if (transactionManager != null && isTransactionActive()) {
+ transactionManager.setRollbackOnly();
+ LOG.debug("Marked transaction for rollback due to exception in
job execution (Quarkus/Narayana)");
+ } else {
+ LOG.trace("No active transaction to mark for rollback");
+ }
+ } catch (SystemException e) {
+ LOG.warn("Failed to mark transaction for rollback", e);
+ }
+ }
+
+ @Override
+ public boolean isTransactionActive() {
+ try {
+ if (transactionManager != null) {
+ int status = transactionManager.getStatus();
+ return status == Status.STATUS_ACTIVE || status ==
Status.STATUS_MARKED_ROLLBACK;
+ }
+ } catch (SystemException e) {
+ LOG.trace("Failed to check transaction status", e);
+ }
+ return false;
+ }
+}
diff --git
a/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/MockDataRepository.java
b/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/MockDataRepository.java
new file mode 100644
index 000000000..db04c5d11
--- /dev/null
+++
b/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/MockDataRepository.java
@@ -0,0 +1,115 @@
+/*
+ * 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.app.jobs.jpa.quarkus;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.transaction.Status;
+import jakarta.transaction.Synchronization;
+import jakarta.transaction.TransactionManager;
+
+/**
+ * Mock repository that tracks data persistence attempts and simulates
+ * transactional behavior to verify that rollbacks prevent duplicate data.
+ */
+@ApplicationScoped
+public class MockDataRepository {
+
+ private final ConcurrentHashMap<String, AtomicInteger> committedData = new
ConcurrentHashMap<>();
+ private final AtomicInteger totalPersistenceAttempts = new
AtomicInteger(0);
+ private final AtomicInteger successfulCommits = new AtomicInteger(0);
+ private final AtomicInteger rolledBackAttempts = new AtomicInteger(0);
+
+ private TransactionManager transactionManager;
+
+ public void setTransactionManager(TransactionManager transactionManager) {
+ this.transactionManager = transactionManager;
+ }
+
+ /**
+ * Simulates persisting data within a transaction.
+ * If the transaction is rolled back, the data should not be committed.
+ */
+ public void persistData(String key, String value) {
+ totalPersistenceAttempts.incrementAndGet();
+
+ // Register a transaction synchronization to track commit/rollback
+ if (transactionManager != null) {
+ try {
+ if (transactionManager.getStatus() == Status.STATUS_ACTIVE) {
+
transactionManager.getTransaction().registerSynchronization(new
Synchronization() {
+ @Override
+ public void beforeCompletion() {
+ // Nothing to do before completion
+ }
+
+ @Override
+ public void afterCompletion(int status) {
+ if (status == Status.STATUS_COMMITTED) {
+ // Only commit the data if transaction succeeds
+ committedData.computeIfAbsent(key, k -> new
AtomicInteger(0)).incrementAndGet();
+ successfulCommits.incrementAndGet();
+ } else if (status == Status.STATUS_ROLLEDBACK) {
+ rolledBackAttempts.incrementAndGet();
+ }
+ }
+ });
+ } else {
+ // No active transaction - commit immediately (shouldn't
happen in our tests)
+ committedData.computeIfAbsent(key, k -> new
AtomicInteger(0)).incrementAndGet();
+ successfulCommits.incrementAndGet();
+ }
+ } catch (Exception e) {
+ // If we can't register synchronization, commit immediately
+ committedData.computeIfAbsent(key, k -> new
AtomicInteger(0)).incrementAndGet();
+ successfulCommits.incrementAndGet();
+ }
+ } else {
+ // No transaction manager - commit immediately
+ committedData.computeIfAbsent(key, k -> new
AtomicInteger(0)).incrementAndGet();
+ successfulCommits.incrementAndGet();
+ }
+ }
+
+ public int getCommittedCount(String key) {
+ AtomicInteger count = committedData.get(key);
+ return count != null ? count.get() : 0;
+ }
+
+ public int getTotalPersistenceAttempts() {
+ return totalPersistenceAttempts.get();
+ }
+
+ public int getSuccessfulCommits() {
+ return successfulCommits.get();
+ }
+
+ public int getRolledBackAttempts() {
+ return rolledBackAttempts.get();
+ }
+
+ public void reset() {
+ committedData.clear();
+ totalPersistenceAttempts.set(0);
+ successfulCommits.set(0);
+ rolledBackAttempts.set(0);
+ }
+}
diff --git
a/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/QuarkusTransactionRollbackTest.java
b/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/QuarkusTransactionRollbackTest.java
new file mode 100644
index 000000000..26ad0f36f
--- /dev/null
+++
b/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/QuarkusTransactionRollbackTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.app.jobs.jpa.quarkus;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.kie.kogito.app.jobs.quarkus.QuarkusJobsService;
+import org.kie.kogito.app.jobs.quarkus.QuarkusTransactionRollbackMarker;
+import org.kie.kogito.jobs.ExactExpirationTime;
+import org.kie.kogito.jobs.JobsService;
+import org.kie.kogito.jobs.descriptors.ProcessInstanceJobDescription;
+
+import io.quarkus.test.junit.QuarkusTest;
+
+import jakarta.inject.Inject;
+import jakarta.transaction.Status;
+import jakarta.transaction.TransactionManager;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@QuarkusTest
+public class QuarkusTransactionRollbackTest {
+
+ @Inject
+ JobsService jobsService;
+
+ @Inject
+ TestJobSchedulerListener listener;
+
+ @Inject
+ TestJobExecutor testJobExecutor;
+
+ @Inject
+ TestExceptionHandler exceptionHandler;
+
+ @Inject
+ QuarkusTransactionRollbackMarker transactionRollbackMarker;
+
+ @Inject
+ TransactionManager transactionManager;
+
+ @Inject
+ MockDataRepository mockDataRepository;
+
+ @BeforeEach
+ public void init() {
+ ((QuarkusJobsService) jobsService).init();
+ testJobExecutor.reset();
+ exceptionHandler.reset();
+ mockDataRepository.reset();
+ mockDataRepository.setTransactionManager(transactionManager);
+ }
+
+ @AfterEach
+ public void cleanup() {
+ ((QuarkusJobsService) jobsService).destroy();
+ }
+
+ @Test
+ public void testTransactionRollbackMarkerInjected() {
+ assertThat(transactionRollbackMarker).isNotNull();
+ assertThat(transactionRollbackMarker.isTransactionActive()).isFalse();
+ }
+
+ @Test
+ public void testTransactionRollbackOnJobFailure() throws Exception {
+ // Configure the test executor to fail 4 times (initial + 3 retries)
+ testJobExecutor.setNumberOfFailures(4);
+
+ ProcessInstanceJobDescription jobDescription = new
ProcessInstanceJobDescription(
+ "rollback-test-job",
+ "-1",
+
ExactExpirationTime.of(Instant.now().plus(Duration.ofSeconds(2)).atZone(ZoneId.of("UTC"))),
+ 5,
+ "test-process-instance",
+ null,
+ "test-process",
+ null,
+ "test-node-instance");
+
+ listener.setCount(4);
+ jobsService.scheduleJob(jobDescription);
+
+ // Wait for all retries to complete and error handler to be invoked
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(10L))
+ .untilAsserted(() -> {
+ assertThat(exceptionHandler.isError())
+ .as("Error handler should be invoked after max
retries")
+ .isTrue();
+ });
+
+ assertThat(listener.await(1, java.util.concurrent.TimeUnit.SECONDS))
+ .as("All job executions should have been attempted")
+ .isTrue();
+
+ assertThat(mockDataRepository.getTotalPersistenceAttempts())
+ .as("Should have 4 persistence attempts (initial + 3 retries)")
+ .isEqualTo(4);
+
+ assertThat(mockDataRepository.getRolledBackAttempts())
+ .as("All failed attempts should have been rolled back")
+ .isEqualTo(4);
+
+ assertThat(mockDataRepository.getSuccessfulCommits())
+ .as("No data should be committed when all attempts fail")
+ .isEqualTo(0);
+
+ assertThat(mockDataRepository.getCommittedCount("rollback-test-job"))
+ .as("No duplicate data should be persisted due to rollback")
+ .isEqualTo(0);
+ }
+
+ @Test
+ public void testTransactionRollbackMarkerFunctionality() throws Exception {
+ assertThat(transactionRollbackMarker.isTransactionActive()).isFalse();
+
+ transactionManager.begin();
+ try {
+
assertThat(transactionRollbackMarker.isTransactionActive()).isTrue();
+
+ transactionRollbackMarker.markForRollback();
+
+ // Verify the transaction is marked for rollback
+ assertThat(transactionManager.getStatus())
+ .as("Transaction should be marked for rollback")
+ .isEqualTo(Status.STATUS_MARKED_ROLLBACK);
+ } finally {
+ transactionManager.rollback();
+ }
+
+ assertThat(transactionRollbackMarker.isTransactionActive()).isFalse();
+ }
+
+}
diff --git
a/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/TestJobExecutor.java
b/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/TestJobExecutor.java
index 8cec79896..e911fc590 100644
---
a/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/TestJobExecutor.java
+++
b/kogito-apps-quarkus/jobs-quarkus/kogito-addons-quarkus-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/jpa/quarkus/TestJobExecutor.java
@@ -23,6 +23,7 @@ import org.kie.kogito.jobs.service.model.JobDetails;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import jakarta.inject.Inject;
import jakarta.inject.Singleton;
@Singleton
@@ -31,6 +32,9 @@ public class TestJobExecutor implements JobExecutor {
private Logger LOG = LoggerFactory.getLogger(TestJobExecutor.class);
private int numberOfFailures;
+ @Inject
+ MockDataRepository mockDataRepository;
+
@Override
public boolean accept(JobDetails jobDescription) {
return true;
@@ -39,6 +43,12 @@ public class TestJobExecutor implements JobExecutor {
@Override
public void execute(JobDetails jobDescription) {
LOG.info("executing {}", jobDescription);
+
+ // Simulate data persistence (e.g., creating a user task)
+ if (mockDataRepository != null) {
+ mockDataRepository.persistData(jobDescription.getId(),
"execution-data");
+ }
+
if (numberOfFailures > 0) {
--numberOfFailures;
throw new RuntimeException();
diff --git
a/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/SpringBootTransactionRollbackMarker.java
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/SpringBootTransactionRollbackMarker.java
new file mode 100644
index 000000000..9c44b3f9b
--- /dev/null
+++
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/main/java/org/kie/kogito/app/jobs/springboot/SpringBootTransactionRollbackMarker.java
@@ -0,0 +1,61 @@
+/*
+ * 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.app.jobs.springboot;
+
+import org.kie.kogito.app.jobs.spi.TransactionRollbackMarker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.TransactionStatus;
+import
org.springframework.transaction.support.TransactionSynchronizationManager;
+
+@Component
+public class SpringBootTransactionRollbackMarker implements
TransactionRollbackMarker {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SpringBootTransactionRollbackMarker.class);
+
+ // Thread-local to store the current TransactionStatus for
TransactionTemplate usage
+ private static final ThreadLocal<TransactionStatus>
CURRENT_TRANSACTION_STATUS = new ThreadLocal<>();
+
+ public static void setCurrentTransactionStatus(TransactionStatus status) {
+ CURRENT_TRANSACTION_STATUS.set(status);
+ }
+
+ public static void clearCurrentTransactionStatus() {
+ CURRENT_TRANSACTION_STATUS.remove();
+ }
+
+ @Override
+ public void markForRollback() {
+ if (isTransactionActive()) {
+ TransactionStatus status = CURRENT_TRANSACTION_STATUS.get();
+ if (status != null) {
+ status.setRollbackOnly();
+ LOG.debug("Transaction marked for rollback using
TransactionStatus");
+ } else {
+ LOG.warn("No TransactionStatus available to mark for
rollback");
+ }
+ }
+ }
+
+ @Override
+ public boolean isTransactionActive() {
+ return TransactionSynchronizationManager.isActualTransactionActive();
+ }
+}
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 47c42cbea..b71251990 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
@@ -32,6 +32,7 @@ import
org.kie.kogito.app.jobs.integrations.ProcessJobDescriptionJobInstanceEven
import
org.kie.kogito.app.jobs.integrations.UserTaskInstanceJobDescriptionJobInstanceEventAdapter;
import org.kie.kogito.app.jobs.spi.JobContextFactory;
import org.kie.kogito.app.jobs.spi.JobStore;
+import org.kie.kogito.app.jobs.spi.TransactionRollbackMarker;
import org.kie.kogito.app.jobs.springboot.resource.RestApiConstants;
import org.kie.kogito.event.EventPublisher;
import org.kie.kogito.handler.ExceptionHandler;
@@ -93,6 +94,9 @@ public class SpringbootJobsService implements JobsService {
@Autowired
protected WrappingConditionalJobExceptionDetailsExtractor
exceptionDetailsExtractor;
+ @Autowired
+ protected TransactionRollbackMarker transactionRollbackMarker;
+
@PostConstruct
public void init() {
this.jobScheduler = JobSchedulerBuilder.newJobSchedulerBuilder()
@@ -110,9 +114,10 @@ public class SpringbootJobsService implements JobsService {
.withMaxNumberOfRetries(maxNumberOfRetries)
.withRefreshJobsInterval(maxRefreshJobsIntervalWindow * 60 *
1000L)
.withTimeoutInterceptor(
- new
TransactionJobTimeoutInterceptor(transactionManager),
- new
ErrorHandlingJobTimeoutInterceptor(ofNullable(exceptionHandlers).stream().toList()))
+ new
ErrorHandlingJobTimeoutInterceptor(ofNullable(exceptionHandlers).stream().toList()),
+ new
TransactionJobTimeoutInterceptor(transactionManager))
.withExceptionDetailsExtractor(exceptionDetailsExtractor)
+ .withTransactionRollbackMarker(transactionRollbackMarker)
.withNumberOfWorkerThreads(numberOfWorkerThreads)
.withJobSynchronization(new JobSynchronization() {
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 58d4b562d..f2ef72f08 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,6 +23,7 @@ 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.support.TransactionTemplate;
public class TransactionJobTimeoutInterceptor implements JobTimeoutInterceptor
{
@@ -41,9 +42,25 @@ public class TransactionJobTimeoutInterceptor implements
JobTimeoutInterceptor {
public JobTimeoutExecution call() throws Exception {
return transactionTemplate.execute(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-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/springboot/MockDataRepository.java
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/springboot/MockDataRepository.java
new file mode 100644
index 000000000..e08991d75
--- /dev/null
+++
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/springboot/MockDataRepository.java
@@ -0,0 +1,98 @@
+/*
+ * 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.app.jobs.springboot;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.support.TransactionSynchronization;
+import
org.springframework.transaction.support.TransactionSynchronizationManager;
+
+/**
+ * Mock repository that tracks data persistence attempts and simulates
+ * transactional behavior to verify that rollbacks prevent duplicate data.
+ */
+@Component
+public class MockDataRepository {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MockDataRepository.class);
+
+ private final ConcurrentHashMap<String, AtomicInteger> committedData = new
ConcurrentHashMap<>();
+ private final AtomicInteger totalPersistenceAttempts = new
AtomicInteger(0);
+ private final AtomicInteger successfulCommits = new AtomicInteger(0);
+ private final AtomicInteger rolledBackAttempts = new AtomicInteger(0);
+
+ /**
+ * Simulates persisting data within a transaction.
+ * If the transaction is rolled back, the data should not be committed.
+ */
+ public void persistData(String key, String value) {
+ totalPersistenceAttempts.incrementAndGet();
+
+ // Register a transaction synchronization to track commit/rollback
+ if (TransactionSynchronizationManager.isSynchronizationActive()) {
+ TransactionSynchronizationManager.registerSynchronization(new
TransactionSynchronization() {
+ @Override
+ public void afterCommit() {
+ // Only commit the data if transaction succeeds
+ committedData.computeIfAbsent(key, k -> new
AtomicInteger(0)).incrementAndGet();
+ successfulCommits.incrementAndGet();
+ }
+
+ @Override
+ public void afterCompletion(int status) {
+ if (status == STATUS_ROLLED_BACK) {
+ rolledBackAttempts.incrementAndGet();
+ }
+ }
+ });
+ } else {
+ // No transaction active - commit immediately (shouldn't happen in
our tests)
+ committedData.computeIfAbsent(key, k -> new
AtomicInteger(0)).incrementAndGet();
+ successfulCommits.incrementAndGet();
+ }
+ }
+
+ public int getCommittedCount(String key) {
+ AtomicInteger count = committedData.get(key);
+ return count != null ? count.get() : 0;
+ }
+
+ public int getTotalPersistenceAttempts() {
+ return totalPersistenceAttempts.get();
+ }
+
+ public int getSuccessfulCommits() {
+ return successfulCommits.get();
+ }
+
+ public int getRolledBackAttempts() {
+ return rolledBackAttempts.get();
+ }
+
+ public void reset() {
+ committedData.clear();
+ totalPersistenceAttempts.set(0);
+ successfulCommits.set(0);
+ rolledBackAttempts.set(0);
+ }
+}
diff --git
a/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/springboot/SpringBootTransactionRollbackTest.java
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/springboot/SpringBootTransactionRollbackTest.java
new file mode 100644
index 000000000..781bbc0fe
--- /dev/null
+++
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/springboot/SpringBootTransactionRollbackTest.java
@@ -0,0 +1,122 @@
+/*
+ * 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.app.jobs.springboot;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.kie.kogito.jobs.ExactExpirationTime;
+import org.kie.kogito.jobs.JobsService;
+import org.kie.kogito.jobs.descriptors.ProcessInstanceJobDescription;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.transaction.PlatformTransactionManager;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest
+public class SpringBootTransactionRollbackTest {
+
+ @Autowired
+ JobsService jobsService;
+
+ @Autowired
+ TestJobSchedulerListener listener;
+
+ @Autowired
+ TestJobExecutor testJobExecutor;
+
+ @Autowired
+ TestExceptionHandler exceptionHandler;
+
+ @Autowired
+ SpringBootTransactionRollbackMarker transactionRollbackMarker;
+
+ @Autowired
+ PlatformTransactionManager transactionManager;
+
+ @Autowired
+ MockDataRepository mockDataRepository;
+
+ @BeforeEach
+ public void init() {
+ testJobExecutor.reset();
+ exceptionHandler.reset();
+ mockDataRepository.reset();
+ }
+
+ @Test
+ public void testTransactionRollbackMarkerInjected() {
+ assertThat(transactionRollbackMarker).isNotNull();
+ assertThat(transactionRollbackMarker.isTransactionActive()).isFalse();
+ }
+
+ @Test
+ public void testTransactionRollbackOnJobFailure() throws Exception {
+ // Configure the test executor to fail 4 times (initial + 3 retries)
+ testJobExecutor.setNumberOfFailures(4);
+
+ ProcessInstanceJobDescription jobDescription = new
ProcessInstanceJobDescription(
+ "rollback-test-job",
+ "-1",
+
ExactExpirationTime.of(Instant.now().plus(Duration.ofSeconds(2)).atZone(ZoneId.of("UTC"))),
+ 5,
+ "test-process-instance",
+ null,
+ "test-process",
+ null,
+ "test-node-instance");
+
+ listener.setCount(4);
+ jobsService.scheduleJob(jobDescription);
+
+ // Wait for all retries to complete and error handler to be invoked
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(10L))
+ .untilAsserted(() -> {
+ assertThat(exceptionHandler.isError())
+ .as("Error handler should be invoked after max
retries")
+ .isTrue();
+ });
+
+ assertThat(listener.await(1, java.util.concurrent.TimeUnit.SECONDS))
+ .as("All job executions should have been attempted")
+ .isTrue();
+
+ assertThat(mockDataRepository.getTotalPersistenceAttempts())
+ .as("Should have 4 persistence attempts (initial + 3 retries)")
+ .isEqualTo(4);
+
+ assertThat(mockDataRepository.getRolledBackAttempts())
+ .as("All failed attempts should have been rolled back by
Spring")
+ .isEqualTo(4);
+
+ assertThat(mockDataRepository.getSuccessfulCommits())
+ .as("No data should be committed when all attempts fail")
+ .isEqualTo(0);
+
+ assertThat(mockDataRepository.getCommittedCount("rollback-test-job"))
+ .as("No duplicate data should be persisted due to automatic
rollback")
+ .isEqualTo(0);
+ }
+}
diff --git
a/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/springboot/TestJobExecutor.java
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/springboot/TestJobExecutor.java
index 52ee0627b..5f08e03b4 100644
---
a/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/springboot/TestJobExecutor.java
+++
b/kogito-apps-springboot/jobs-springboot/kogito-addons-springboot-embedded-jobs/src/test/java/org/kie/kogito/app/jobs/springboot/TestJobExecutor.java
@@ -22,6 +22,7 @@ import org.kie.kogito.app.jobs.api.JobExecutor;
import org.kie.kogito.jobs.service.model.JobDetails;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
@@ -33,6 +34,9 @@ public class TestJobExecutor implements JobExecutor {
private Logger LOG = LoggerFactory.getLogger(TestJobExecutor.class);
private int numberOfFailures;
+ @Autowired(required = false)
+ private MockDataRepository mockDataRepository;
+
@Override
public boolean accept(JobDetails jobDescription) {
return true;
@@ -41,6 +45,12 @@ public class TestJobExecutor implements JobExecutor {
@Override
public void execute(JobDetails jobDescription) {
LOG.info("executing {}", jobDescription);
+
+ // Simulate data persistence (e.g., creating a user task)
+ if (mockDataRepository != null) {
+ mockDataRepository.persistData(jobDescription.getId(),
"execution-data");
+ }
+
if (numberOfFailures > 0) {
--numberOfFailures;
throw new RuntimeException();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
