martinweiler commented on code in PR #2333:
URL:
https://github.com/apache/incubator-kie-kogito-apps/pull/2333#discussion_r3500469485
##########
jobs/jobs-common-embedded/src/main/java/org/kie/kogito/app/jobs/impl/VertxJobScheduler.java:
##########
@@ -467,63 +464,38 @@ private void scheduleRetry(JobContext jobContext,
JobDetails jobDetails) {
// 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);
- vertx.executeBlocking(promise -> {
- try {
- JobContext newJobContext = jobContextFactory.newContext();
-
- Callable<Void> newTransactionTask = () -> {
- jobStore.update(newJobContext, jobDetails);
- fireEvents(jobDetails); // Fire events after rollback
- // Schedule timer in new transaction after persistence
- if (status == JobStatus.RETRY) {
- addOrUpdateTxTimer(jobDetails);
- } else {
- removeTxTimer(jobDetails);
- }
- return null;
- };
-
- // Apply transaction interceptors to ensure this runs in a
new transaction
- Callable<Void> wrappedTask = newTransactionTask;
- for (JobTimeoutInterceptor interceptor : interceptors) {
- final Callable<Void> currentTask = wrappedTask;
- wrappedTask = () -> {
- interceptor.chainIntercept(() -> {
- currentTask.call();
- return new JobTimeoutExecution(jobDetails);
- }).call();
- return null;
- };
- }
+ JobContext newJobContext = jobContextFactory.newContext();
- wrappedTask.call();
- promise.complete();
- } catch (Exception e) {
- LOG.error("Failed to schedule retry for job {} in new
transaction", jobId, e);
- promise.fail(e);
- }
- }, false, result -> {
- if (result.failed()) {
- LOG.error("Async retry scheduling failed for job {}",
jobId, result.cause());
+ Callable<JobTimeoutExecution> newTransactionTask = () -> {
+ jobStore.update(newJobContext, jobDetails);
+ fireEvents(jobDetails); // Fire events after rollback
+ // Schedule timer in new transaction after persistence
+ if (status == JobStatus.RETRY) {
+ addOrUpdateTxTimer(jobDetails);
+ } else {
+ removeTxTimer(jobDetails);
}
- });
+ return new JobTimeoutExecution(jobDetails);
+ };
+
+ // 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 {
jobStore.update(jobContext, jobDetails);
- // Fire events AFTER persistence to ensure Data Index can
query the updated job
fireEvents(jobDetails);
- // Schedule timer AFTER persistence (no transaction
synchronization needed)
if (status == JobStatus.RETRY) {
- addTimerInfo(jobDetails);
+ addOrUpdateTxTimer(jobDetails);
} else {
- String mapKey = getMapKey(jobDetails);
- TimerInfo timerInfo = jobsScheduled.remove(mapKey);
- if (timerInfo != null) {
- removeTimerInfo(timerInfo);
- }
+ removeTxTimer(jobDetails);
Review Comment:
Thanks @jstastny-cz ! I moved the duplicate steps into a new
`persistAndSchedule` method, to avoid potential issues in the future in case we
have to change this logic. I did not implement your suggestion of reusing
`newTransactionTask.call()` as it would have complicated the exception
handling.
--
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]
