JSL_Digital_Solution
/
video_editing_poc
forked from jsl/video_editing_poc
2
0
Fork 0
Code Pull Requests Activity

I finished the missing part: retryable DB-queue failures no longer become terminal job failures on the first exception. ClipProcessor now supports non-terminal queue retries, the DB queue adapter reschedules retries and only 

marks the job FAILED when the message actually moves to DLQ, and retry/DLQ events are recorded at the job level. The key changes are in src/main/java/org/example/videoclips/processing/ClipProcessor.java:37 and src/main/java/
  org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapter.java:18.

  I also kept the queue tests aligned in src/test/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapterTest.java:1, reran mvn -q test, and the suite passed:

  - TenantQuotaControllerTest: 1/1
  - VideoAssetControllerTest: 7/7
  - RetentionCleanupJobTest: 2/2
  - DatabaseBackedClipJobQueueAdapterTest: 2/2

  The plan is updated accordingly in docs/video-clipping-service-implementation-plan.md:701, with Retry and DLQ handling now marked complete. The next unfinished step is Full application/problem+json error model.
This commit is contained in:
JSLMPR 2026-07-09 01:10:53 +02:00
parent 1bd879f1e2
commit 1737d8b220
5 changed files with 62 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/video-clipping-service-implementation-plan.md
View File

@ -702,7 +702,7 @@ Keep the domain independent from Spring framework details. Adapters implement st
1. [x] `EXACT` clipping mode.
2. [x] Idempotency keys.
3. [~] Retry and DLQ handling.
3. [x] Retry and DLQ handling.
4. [x] Tenant quotas.
5. [~] Full `application/problem+json` error model.
6. [x] Cleanup lifecycle jobs.

5
src/main/java/org/example/videoclips/domain/ClipJob.java
View File

@ -40,4 +40,9 @@ public record ClipJob(
return new ClipJob(id, assetId, tenantId, status, accuracyMode, segmentDurationSeconds, progressPercent, attemptCount,
createdAt, startedAt, finishedAt, message);
}
public ClipJob withAttemptCount(int newAttemptCount) {
return new ClipJob(id, assetId, tenantId, status, accuracyMode, segmentDurationSeconds, progressPercent, newAttemptCount,
createdAt, startedAt, finishedAt, errorMessage);
}
}

41
src/main/java/org/example/videoclips/processing/ClipProcessor.java
View File

@ -45,6 +45,10 @@ public class ClipProcessor {
}
public void processNow(String jobId) {
processNow(jobId, true);
}
public void processNow(String jobId, boolean persistTerminalFailure) {
Path localInputPath = null;
List<GeneratedClip> generatedClips = List.of();
try {
@ -84,14 +88,8 @@ public class ClipProcessor {
repository.saveAsset(requireAsset(asset.id()).withStatus(VideoAsset.Status.READY));
recordEvent(jobId, "SUCCEEDED", "Generated " + generatedClips.size() + " clips");
} catch (Exception ex) {
ClipJob job = repository.findJobById(jobId);
if (job != null) {
repository.saveJob(job.withStatus(ClipJobStatus.FAILED).withErrorMessage(ex.getMessage()).withFinishedAt(Instant.now()));
VideoAsset asset = repository.findAssetById(job.assetId());
if (asset != null) {
repository.saveAsset(asset.withStatus(VideoAsset.Status.UPLOADED));
}
recordEvent(jobId, "FAILED", ex.getMessage());
if (persistTerminalFailure) {
markTerminalFailure(jobId, requireJob(jobId) == null ? 0 : requireJob(jobId).attemptCount(), ex.getMessage());
}
throw new IllegalStateException("Clip processing failed for job " + jobId, ex);
} finally {
@ -99,6 +97,33 @@ public class ClipProcessor {
}
}
public void markRetryScheduled(String jobId, int attemptCount, Instant nextAttemptAt, String message) {
ClipJob job = repository.findJobById(jobId);
if (job == null) {
return;
}
repository.saveJob(job.withAttemptCount(attemptCount).withStatus(ClipJobStatus.QUEUED).withErrorMessage(message));
VideoAsset asset = repository.findAssetById(job.assetId());
if (asset != null) {
repository.saveAsset(asset.withStatus(VideoAsset.Status.UPLOADED));
}
recordEvent(jobId, "RETRY_SCHEDULED", "Retry " + attemptCount + " scheduled for " + nextAttemptAt);
}
public void markTerminalFailure(String jobId, int attemptCount, String message) {
ClipJob job = repository.findJobById(jobId);
if (job == null) {
return;
}
repository.saveJob(job.withAttemptCount(attemptCount).withStatus(ClipJobStatus.FAILED).withErrorMessage(message).withFinishedAt(Instant.now()));
VideoAsset asset = repository.findAssetById(job.assetId());
if (asset != null) {
repository.saveAsset(asset.withStatus(VideoAsset.Status.UPLOADED));
}
recordEvent(jobId, "DLQ", "Moved to dead-letter queue after " + attemptCount + " attempts");
recordEvent(jobId, "FAILED", message);
}
private ClipJob requireJob(String jobId) {
return repository.findJobById(jobId);
}

16
src/main/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapter.java
View File

@ -1,5 +1,7 @@
package org.example.videoclips.queue;
import org.example.videoclips.application.VideoClippingRepository;
import org.example.videoclips.domain.ClipJob;
import org.example.videoclips.config.VideoClippingProperties;
import org.example.videoclips.processing.ClipProcessor;
import org.example.videoclips.queue.entity.QueueMessageEntity;
@ -18,15 +20,18 @@ import java.util.UUID;
public class DatabaseBackedClipJobQueueAdapter implements ClipJobQueuePort {
private final QueueMessageJpaRepository queueMessageJpaRepository;
private final VideoClippingRepository repository;
private final ClipProcessor clipProcessor;
private final VideoClippingProperties properties;
public DatabaseBackedClipJobQueueAdapter(
QueueMessageJpaRepository queueMessageJpaRepository,
VideoClippingRepository repository,
ClipProcessor clipProcessor,
VideoClippingProperties properties
) {
this.queueMessageJpaRepository = queueMessageJpaRepository;
this.repository = repository;
this.clipProcessor = clipProcessor;
this.properties = properties;
}
@ -58,17 +63,24 @@ public class DatabaseBackedClipJobQueueAdapter implements ClipJobQueuePort {
message.setClaimedAt(now);
message.setAttemptCount(message.getAttemptCount() + 1);
queueMessageJpaRepository.save(message);
ClipJob job = repository.findJobById(message.getJobId());
if (job != null) {
repository.saveJob(job.withAttemptCount(message.getAttemptCount()));
}
try {
clipProcessor.processNow(message.getJobId());
clipProcessor.processNow(message.getJobId(), false);
message.setStatus("COMPLETED");
message.setLastError(null);
} catch (Exception ex) {
message.setLastError(ex.getMessage());
if (message.getAttemptCount() >= properties.getDatabaseQueue().getMaxAttempts()) {
message.setStatus("DLQ");
clipProcessor.markTerminalFailure(message.getJobId(), message.getAttemptCount(), ex.getMessage());
} else {
message.setStatus("PENDING");
message.setAvailableAt(now.plusMillis(properties.getDatabaseQueue().getRetryBackoffMs()));
Instant nextAttemptAt = now.plusMillis(properties.getDatabaseQueue().getRetryBackoffMs());
message.setAvailableAt(nextAttemptAt);
clipProcessor.markRetryScheduled(message.getJobId(), message.getAttemptCount(), nextAttemptAt, ex.getMessage());
}
}
queueMessageJpaRepository.save(message);

13
src/test/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapterTest.java
View File

@ -1,5 +1,6 @@
package org.example.videoclips.queue;
import org.example.videoclips.application.VideoClippingRepository;
import org.example.videoclips.config.VideoClippingProperties;
import org.example.videoclips.processing.ClipProcessor;
import org.example.videoclips.queue.entity.QueueMessageEntity;
@ -23,6 +24,7 @@ class DatabaseBackedClipJobQueueAdapterTest {
@Test
void retriesFailedMessagesWithBackoffBeforeSendingToDlq() {
QueueMessageJpaRepository repository = mock(QueueMessageJpaRepository.class);
VideoClippingRepository videoClippingRepository = mock(VideoClippingRepository.class);
ClipProcessor clipProcessor = mock(ClipProcessor.class);
VideoClippingProperties properties = new VideoClippingProperties();
properties.getDatabaseQueue().setMaxAttempts(3);
@ -37,9 +39,9 @@ class DatabaseBackedClipJobQueueAdapterTest {
when(repository.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(any(), any()))
.thenReturn(List.of(firstAttempt));
doThrow(new IllegalStateException("processing failed")).when(clipProcessor).processNow("job_1");
doThrow(new IllegalStateException("processing failed")).when(clipProcessor).processNow("job_1", false);
DatabaseBackedClipJobQueueAdapter adapter = new DatabaseBackedClipJobQueueAdapter(repository, clipProcessor, properties);
DatabaseBackedClipJobQueueAdapter adapter = new DatabaseBackedClipJobQueueAdapter(repository, videoClippingRepository, clipProcessor, properties);
adapter.poll();
@ -48,11 +50,13 @@ class DatabaseBackedClipJobQueueAdapterTest {
assertEquals("processing failed", firstAttempt.getLastError());
assertNotNull(firstAttempt.getClaimedAt());
verify(repository, times(2)).save(firstAttempt);
verify(clipProcessor).markRetryScheduled(any(), any(Integer.class), any(), any());
}
@Test
void movesMessageToDlqAfterMaxAttempts() {
QueueMessageJpaRepository repository = mock(QueueMessageJpaRepository.class);
VideoClippingRepository videoClippingRepository = mock(VideoClippingRepository.class);
ClipProcessor clipProcessor = mock(ClipProcessor.class);
VideoClippingProperties properties = new VideoClippingProperties();
properties.getDatabaseQueue().setMaxAttempts(3);
@ -67,9 +71,9 @@ class DatabaseBackedClipJobQueueAdapterTest {
when(repository.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(any(), any()))
.thenReturn(List.of(finalAttempt));
doThrow(new IllegalStateException("processing failed")).when(clipProcessor).processNow("job_2");
doThrow(new IllegalStateException("processing failed")).when(clipProcessor).processNow("job_2", false);
DatabaseBackedClipJobQueueAdapter adapter = new DatabaseBackedClipJobQueueAdapter(repository, clipProcessor, properties);
DatabaseBackedClipJobQueueAdapter adapter = new DatabaseBackedClipJobQueueAdapter(repository, videoClippingRepository, clipProcessor, properties);
adapter.poll();
@ -77,5 +81,6 @@ class DatabaseBackedClipJobQueueAdapterTest {
assertEquals("DLQ", finalAttempt.getStatus());
assertEquals("processing failed", finalAttempt.getLastError());
verify(repository, times(2)).save(finalAttempt);
verify(clipProcessor).markTerminalFailure("job_2", 3, "processing failed");
}
}