jstastny-cz commented on code in PR #2333:
URL:
https://github.com/apache/incubator-kie-kogito-apps/pull/2333#discussion_r3474455093
##########
jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java:
##########
@@ -448,12 +519,36 @@ private void reconcileScheduling(Long timerId, JobContext
jobContext, JobDetails
case RETRY:
LOG.trace("Timeout {} with jobId {} will be updated and
scheduled", timerId, jobId);
addOrUpdateTxTimer(nextJobDetails);
- jobStore.update(jobContext, nextJobDetails);
+ // Use persist instead of update to handle case where
transaction was rolled back
+ // and job record no longer exists in database
+ JobDetails existing = jobStore.find(jobContext, jobId);
+ if (existing != null) {
+ jobStore.update(jobContext, nextJobDetails);
Review Comment:
just FYI, update is implemented as merge.
##########
jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java:
##########
@@ -416,14 +427,29 @@ public JobTimeoutExecution call() throws Exception {
JobDetails executeJobDetails =
doExecute(runningJobDetails);
LOG.trace("Timeout {} with jobId {} will be rescheduled if
required", timerId, jobId);
JobDetails nextJobDetails =
computeAndScheduleNextJobIfAny(executeJobDetails);
- reconcileScheduling(timerId, jobContext, nextJobDetails);
+ reconcileScheduling(timerId, jobContext, nextJobDetails,
false);
jobSchedulerListeners.forEach(l ->
l.onExecution(jobDetails));
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
+ markTransactionForRollback();
+
JobDetails nextJobDetails = doRetryOrError(jobDetails,
exception);
- reconcileScheduling(timerId, jobContext, nextJobDetails);
+
+ // For RETRY and ERROR status, check if we need to
reconcile in a NEW transaction
+ // Only do this if there's an active transaction that was
marked for rollback
+ if ((nextJobDetails.getStatus() == JobStatus.RETRY ||
nextJobDetails.getStatus() == JobStatus.ERROR)
Review Comment:
I'd suggest creating boolean method named accordingly instead of complex if
conditions and additional comments.
##########
jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java:
##########
@@ -416,14 +427,29 @@ public JobTimeoutExecution call() throws Exception {
JobDetails executeJobDetails =
doExecute(runningJobDetails);
LOG.trace("Timeout {} with jobId {} will be rescheduled if
required", timerId, jobId);
JobDetails nextJobDetails =
computeAndScheduleNextJobIfAny(executeJobDetails);
- reconcileScheduling(timerId, jobContext, nextJobDetails);
+ reconcileScheduling(timerId, jobContext, nextJobDetails,
false);
jobSchedulerListeners.forEach(l ->
l.onExecution(jobDetails));
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
+ markTransactionForRollback();
Review Comment:
```suggestion
markTransactionForRollbackWhenEnabled();
```
or
```suggestion
markTransactionForRollbackWhenConfigured();
```
##########
jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java:
##########
@@ -435,7 +461,52 @@ public JobTimeoutExecution call() throws Exception {
return current;
}
- private void reconcileScheduling(Long timerId, JobContext jobContext,
JobDetails nextJobDetails) {
+ /**
+ * Schedules reconciliation to happen in a new transaction after the
current one completes.
+ * This is necessary when the current transaction is marked for rollback
due to an exception.
+ */
+ private void scheduleReconciliationInNewTransaction(Long timerId,
JobDetails jobDetails) {
+ String jobId = jobDetails.getId();
+
+ // Schedule the reconciliation to run asynchronously in a new
transaction
+ // This must be async to avoid deadlock with the current transaction
+ vertx.executeBlocking(promise -> {
+ try {
+ JobContext newJobContext = jobContextFactory.newContext();
+
+ Callable<Void> reconciliationTask = () -> {
+ reconcileScheduling(timerId, newJobContext, jobDetails,
true);
+ return null;
+ };
+
+ // Apply transaction interceptors to ensure this runs in a new
transaction
+ Callable<Void> wrappedTask = reconciliationTask;
+ for (JobTimeoutInterceptor interceptor : interceptors) {
+ final Callable<Void> currentTask = wrappedTask;
+ wrappedTask = () -> {
+ interceptor.chainIntercept(() -> {
+ currentTask.call();
+ return new JobTimeoutExecution(jobDetails);
+ }).call();
+ return null;
+ };
+ }
+
+ wrappedTask.call();
+
+ promise.complete();
+ } catch (Exception e) {
+ LOG.error("Failed to reconcile job {} in new transaction",
jobId, e);
+ promise.fail(e);
+ }
+ }, false, result -> {
+ if (result.failed()) {
+ LOG.error("Async reconciliation failed for job {}", jobId,
result.cause());
+ }
+ });
+ }
+
+ private void reconcileScheduling(Long timerId, JobContext jobContext,
JobDetails nextJobDetails, boolean afterRollback) {
Review Comment:
why not switch the logic and call `scheduleReconciliationInNewTransaction`
from here - it would improve encapsulation and reconcileScheduling would still
be the main place to tweak scheduling. We're again spreading the scheduling all
over the place.
scheduleReconciliationInNewTransaction is not doing the reconciliation
anyways, because it is only invoked for subset of states, so IMHO the semantics
don't match.
perhaps we need an operation like "scheduleRetry" which would be invoked in
both `RETRY` and `ERROR` case in reconcileScheduling switch(they're copy&paste
anyways already). Then inside this new method, we'd check if we have the valid
case:
```
(nextJobDetails.getStatus() == JobStatus.RETRY || nextJobDetails.getStatus()
== JobStatus.ERROR) && transactionRollbackMarker != null &&
transactionRollbackMarker.isTransactionActive()
```
Now, the JobStatus check would be redundant in the call stack, so much
simpler:
```
transactionRollbackMarker != null &&
transactionRollbackMarker.isTransactionActive()
```
If the condition is true, we'd need to handle the vertx new transaction
scheduling, if not, we simply run what we need to. Even further - "what we need
to" is your original
```
JobDetails existingError = jobStore.find(jobContext, jobId);
if (existingError != null) {
jobStore.update(jobContext, nextJobDetails);
} else {
jobStore.persist(jobContext, nextJobDetails);
}
// Fire events only if we're reconciling after a rollback
// (events from original transaction were lost)
if (afterRollback) {
fireEvents(nextJobDetails);
}
```
And the `if (afterRollback)` part would then be obvious from within the
`scheduleRetry`, right? Based on `transactionRollbackMarker != null &&
transactionRollbackMarker.isTransactionActive()` again.
So in my naive perspective
```
Callable<void> scheduleRetryCall = () -> {
JobDetails existingError = jobStore.find(jobContext, jobId);
if (existingError != null) {
jobStore.update(jobContext, nextJobDetails);
} else {
jobStore.persist(jobContext, nextJobDetails);
}
};
if(transactionRollbackMarker != null &&
transactionRollbackMarker.isTransactionActive()) {
// vertx.executeBlocking with wrap callable from
scheduleRetryCall var above.
fireEvents(nextJobDetails); // this now is needed
unconditionally, right?
} else {
scheduleRetryCall.call();
}
}
```
Does it make sense?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
