Keep row locked when exception occurs inside transaction - Spring Boot

Viewed 39

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:

  1. Change the runInTransaction to @Transactional(noRollbackFor = RuntimeException.class);
  2. Add @Transactional(propagation = Propagation.REQUIRES_NEW) on the rescheduleJobExecution method. 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.

0 Answers
Related