We are having trouble to keep a transaction running when a exception occurs and also keep rows locked when this happens.
Firs, a bit of code to clarify.
With this query we are performing a for update skip locked (JobDao):
@NotNull
default Optional<EventEntity> getJobForProcessing(@NotNull final Instant now) {
final List<EventEntity> events = getUnprocessedEvent(now, PageRequest.of(0, 1));
if (events.isEmpty()) {
return Optional.empty();
}
return Optional.ofNullable(events.get(0));
}
@NotNull
@Query("""
select ee from EventEntity ee
where ee.processed = false
order by ee.createdAt""")
@Lock(LockModeType.PESSIMISTIC_WRITE)
@QueryHints({@QueryHint(name = "javax.persistence.lock.timeout", value = LockOptions.SKIP_LOCKED + "")})
Optional<EventEntity> getJobForProcessing(@NotNull final Instant now, @NotNull final Pageable pageable);
The following code runs on another thread using @Async (JobTransaction called from scheduler):
@Transactional
public void runInTransaction() {
final Instant now = Now.getInstantUtc();
final Optional<Job> jobForProcessing = dao.getJobForProcessing(now);
if (jobForProcessing.isPresent()) {
final Job job = jobForProcessing.get();
jobProcessor.processJob(job);
} else {
Log.d(TAG, "No more jobs, stopping thread %s.".formatted(Thread.currentThread().getName()));
}
}
The processJob(job) method — simplified — is the following (JobProcessor):
private void processJob(@NotNull final Job job) {
try {
worker.execute(job);
jobService.markJobAsProcessed(job);
} catch (final Throwable throwable) {
Log.e(TAG, "Error to process job: " + job.getId(), throwable);
jobService.rescheduleJobExecution(job, throwable);
throw new RuntimeException(throwable);
}
}
On the above code, the important part is the catch block that calls jobService.rescheduleJobExecution(...).
This is the rescheduleJobExecution (JobService):
public void rescheduleJobExecution(@NotNull final Job job,
@NotNull final Throwable throwable) {
final var nextRetryAt = backoffCalculator.calculateNextRetryAt(job, now);
final var stackTraceAsString = Throwables.getStackTraceAsString(throwable);
final var executionAttempts = job.getExecutionAttempts() + 1;
final var newJob = job.cloneEntity()
.withExecutionAttempts(executionAttempts)
.withErrorLogLastRetry(stackTraceAsString)
.withNextRetryAt(nextRetryAt)
.build();
jobDao.save(newJob);
}
The problem I'm facing is that, when one error occurs and we try to reschedule the job, the following is throw: Transaction silently rolled back because it has been marked as rollback-only.
I understand why this happens, but have not found any way to fix this. What I already tried:
- Change the
runInTransactionto@Transactional(noRollbackFor = RuntimeException.class); - Add
@Transactional(propagation = Propagation.REQUIRES_NEW)on therescheduleJobExecutionmethod. This solves the problem but brings a new one: since the outer transaction has already failed, the row is available to other threads to select and in a multithread environment this could lead to race conditions (ie: the job is reselected to execution before the reschedule is done).
So, I'm looking for a solution that enables me to deal with job rescheduling and also that keep jobs safe on a multithread environment. The getUnprocessedEvent method is used by many threads and we also like to keep using the for update skip locked that postgres provides.