jstastny-cz commented on code in PR #2333:
URL:
https://github.com/apache/incubator-kie-kogito-apps/pull/2333#discussion_r3489959308
##########
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:
What I actually meant was, reuse the `newTransactionTask` callable in here
(after moved to higher level context and renamed perhaps), so that we define
this list of steps in a single place?
So in here I'd expect sth like `newTransactionTask.call()`; What do you
think? Would we have problems catching the exception in such case? Would it
then be more appropriate to create a reusable method out of
```
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);
```
and call it both in here an in the `newTransactionTask` ?
I mean, the logic of diverting the execution based on if status is RETRY or
not, makes the procedure non-trivial, so my worry is if we'd update
appropriately in both places in future.
--
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]
