• I updated the milestone section in docs/video-clipping-service-implementation-plan.md:688 to mark what is already done and to show Retry and DLQ handling as in progress rather than untouched.
I also started that next step in code. The database-backed queue now retries failed jobs with configurable backoff and moves messages to a terminal DLQ status after the configured max attempts in src/main/java/org/example/ videoclips/queue/DatabaseBackedClipJobQueueAdapter.java:48. The new queue settings live in src/main/java/org/example/videoclips/config/VideoClippingProperties.java:133, and I added focused coverage in src/test/java/org/ example/videoclips/queue/DatabaseBackedClipJobQueueAdapterTest.java:1. Verification: mvn -q test passed. - VideoAssetControllerTest: 7/7 - RetentionCleanupJobTest: 2/2 - DatabaseBackedClipJobQueueAdapterTest: 2/2
This commit is contained in:
JSLMPR
parent
0d45f898f7
commit
8548542248
|
|
@ -689,42 +689,42 @@ Keep the domain independent from Spring framework details. Adapters implement st
|
|||
|
||||
### MVP
|
||||
|
||||
1. Create asset and upload-session API.
|
||||
2. Direct S3 multipart upload support.
|
||||
3. Complete upload endpoint.
|
||||
4. Create clip job endpoint.
|
||||
5. Queue publisher and single worker.
|
||||
6. FFmpeg `FAST` mode.
|
||||
7. Job status and clip listing.
|
||||
8. Basic integration tests.
|
||||
1. [x] Create asset and upload-session API.
|
||||
2. [x] Direct S3 multipart upload support.
|
||||
3. [x] Complete upload endpoint.
|
||||
4. [x] Create clip job endpoint.
|
||||
5. [x] Queue publisher and single worker.
|
||||
6. [x] FFmpeg `FAST` mode.
|
||||
7. [x] Job status and clip listing.
|
||||
8. [x] Basic integration tests.
|
||||
|
||||
### Production Hardening
|
||||
|
||||
1. `EXACT` clipping mode.
|
||||
2. Idempotency keys.
|
||||
3. Retry and DLQ handling.
|
||||
4. Tenant quotas.
|
||||
5. Full `application/problem+json` error model.
|
||||
6. Cleanup lifecycle jobs.
|
||||
7. Security review.
|
||||
1. [x] `EXACT` clipping mode.
|
||||
2. [x] Idempotency keys.
|
||||
3. [~] Retry and DLQ handling.
|
||||
4. [ ] Tenant quotas.
|
||||
5. [~] Full `application/problem+json` error model.
|
||||
6. [x] Cleanup lifecycle jobs.
|
||||
7. [~] Security review.
|
||||
|
||||
### Performance Tuning
|
||||
|
||||
1. Worker CPU/disk benchmarks.
|
||||
2. FFmpeg preset benchmarks.
|
||||
3. Object-storage bandwidth tests.
|
||||
4. Queue visibility-timeout tuning.
|
||||
5. Autoscaling policies.
|
||||
6. Cost model by video minute.
|
||||
1. [ ] Worker CPU/disk benchmarks.
|
||||
2. [ ] FFmpeg preset benchmarks.
|
||||
3. [ ] Object-storage bandwidth tests.
|
||||
4. [ ] Queue visibility-timeout tuning.
|
||||
5. [ ] Autoscaling policies.
|
||||
6. [ ] Cost model by video minute.
|
||||
|
||||
### Operational Readiness
|
||||
|
||||
1. Dashboards.
|
||||
2. Alerts.
|
||||
3. Runbooks.
|
||||
4. Backup and restore validation.
|
||||
5. DLQ redrive procedure.
|
||||
6. Load-test signoff.
|
||||
1. [ ] Dashboards.
|
||||
2. [ ] Alerts.
|
||||
3. [ ] Runbooks.
|
||||
4. [ ] Backup and restore validation.
|
||||
5. [ ] DLQ redrive procedure.
|
||||
6. [ ] Load-test signoff.
|
||||
|
||||
## 14. Open Questions
|
||||
|
||||
|
|
|
|||
|
|
@ -164,6 +164,12 @@ public class VideoClippingProperties {
|
|||
@Min(1)
|
||||
private int batchSize = 5;
|
||||
|
||||
@Min(1)
|
||||
private int maxAttempts = 3;
|
||||
|
||||
@Min(0)
|
||||
private long retryBackoffMs = 5000;
|
||||
|
||||
public long getPollIntervalMs() {
|
||||
return pollIntervalMs;
|
||||
}
|
||||
|
|
@ -179,6 +185,22 @@ public class VideoClippingProperties {
|
|||
public void setBatchSize(int batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
}
|
||||
|
||||
public int getMaxAttempts() {
|
||||
return maxAttempts;
|
||||
}
|
||||
|
||||
public void setMaxAttempts(int maxAttempts) {
|
||||
this.maxAttempts = maxAttempts;
|
||||
}
|
||||
|
||||
public long getRetryBackoffMs() {
|
||||
return retryBackoffMs;
|
||||
}
|
||||
|
||||
public void setRetryBackoffMs(long retryBackoffMs) {
|
||||
this.retryBackoffMs = retryBackoffMs;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Ffmpeg {
|
||||
|
|
|
|||
|
|
@ -47,14 +47,15 @@ public class DatabaseBackedClipJobQueueAdapter implements ClipJobQueuePort {
|
|||
@Transactional
|
||||
public void poll() {
|
||||
int remaining = properties.getDatabaseQueue().getBatchSize();
|
||||
Instant now = Instant.now();
|
||||
List<QueueMessageEntity> messages = queueMessageJpaRepository
|
||||
.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc("PENDING", Instant.now());
|
||||
.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc("PENDING", now);
|
||||
for (QueueMessageEntity message : messages) {
|
||||
if (remaining-- <= 0) {
|
||||
break;
|
||||
}
|
||||
message.setStatus("PROCESSING");
|
||||
message.setClaimedAt(Instant.now());
|
||||
message.setClaimedAt(now);
|
||||
message.setAttemptCount(message.getAttemptCount() + 1);
|
||||
queueMessageJpaRepository.save(message);
|
||||
try {
|
||||
|
|
@ -62,8 +63,13 @@ public class DatabaseBackedClipJobQueueAdapter implements ClipJobQueuePort {
|
|||
message.setStatus("COMPLETED");
|
||||
message.setLastError(null);
|
||||
} catch (Exception ex) {
|
||||
message.setStatus("FAILED");
|
||||
message.setLastError(ex.getMessage());
|
||||
if (message.getAttemptCount() >= properties.getDatabaseQueue().getMaxAttempts()) {
|
||||
message.setStatus("DLQ");
|
||||
} else {
|
||||
message.setStatus("PENDING");
|
||||
message.setAvailableAt(now.plusMillis(properties.getDatabaseQueue().getRetryBackoffMs()));
|
||||
}
|
||||
}
|
||||
queueMessageJpaRepository.save(message);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,81 @@
|
|||
package org.example.videoclips.queue;
|
||||
|
||||
import org.example.videoclips.config.VideoClippingProperties;
|
||||
import org.example.videoclips.processing.ClipProcessor;
|
||||
import org.example.videoclips.queue.entity.QueueMessageEntity;
|
||||
import org.example.videoclips.queue.spring.QueueMessageJpaRepository;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
class DatabaseBackedClipJobQueueAdapterTest {
|
||||
|
||||
@Test
|
||||
void retriesFailedMessagesWithBackoffBeforeSendingToDlq() {
|
||||
QueueMessageJpaRepository repository = mock(QueueMessageJpaRepository.class);
|
||||
ClipProcessor clipProcessor = mock(ClipProcessor.class);
|
||||
VideoClippingProperties properties = new VideoClippingProperties();
|
||||
properties.getDatabaseQueue().setMaxAttempts(3);
|
||||
properties.getDatabaseQueue().setRetryBackoffMs(5000);
|
||||
|
||||
QueueMessageEntity firstAttempt = new QueueMessageEntity();
|
||||
firstAttempt.setId("msg_1");
|
||||
firstAttempt.setJobId("job_1");
|
||||
firstAttempt.setStatus("PENDING");
|
||||
firstAttempt.setAttemptCount(0);
|
||||
firstAttempt.setAvailableAt(Instant.now().minusSeconds(1));
|
||||
|
||||
when(repository.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(any(), any()))
|
||||
.thenReturn(List.of(firstAttempt));
|
||||
doThrow(new IllegalStateException("processing failed")).when(clipProcessor).processNow("job_1");
|
||||
|
||||
DatabaseBackedClipJobQueueAdapter adapter = new DatabaseBackedClipJobQueueAdapter(repository, clipProcessor, properties);
|
||||
|
||||
adapter.poll();
|
||||
|
||||
assertEquals(1, firstAttempt.getAttemptCount());
|
||||
assertEquals("PENDING", firstAttempt.getStatus());
|
||||
assertEquals("processing failed", firstAttempt.getLastError());
|
||||
assertNotNull(firstAttempt.getClaimedAt());
|
||||
verify(repository, times(2)).save(firstAttempt);
|
||||
}
|
||||
|
||||
@Test
|
||||
void movesMessageToDlqAfterMaxAttempts() {
|
||||
QueueMessageJpaRepository repository = mock(QueueMessageJpaRepository.class);
|
||||
ClipProcessor clipProcessor = mock(ClipProcessor.class);
|
||||
VideoClippingProperties properties = new VideoClippingProperties();
|
||||
properties.getDatabaseQueue().setMaxAttempts(3);
|
||||
properties.getDatabaseQueue().setRetryBackoffMs(5000);
|
||||
|
||||
QueueMessageEntity finalAttempt = new QueueMessageEntity();
|
||||
finalAttempt.setId("msg_2");
|
||||
finalAttempt.setJobId("job_2");
|
||||
finalAttempt.setStatus("PENDING");
|
||||
finalAttempt.setAttemptCount(2);
|
||||
finalAttempt.setAvailableAt(Instant.now().minusSeconds(1));
|
||||
|
||||
when(repository.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(any(), any()))
|
||||
.thenReturn(List.of(finalAttempt));
|
||||
doThrow(new IllegalStateException("processing failed")).when(clipProcessor).processNow("job_2");
|
||||
|
||||
DatabaseBackedClipJobQueueAdapter adapter = new DatabaseBackedClipJobQueueAdapter(repository, clipProcessor, properties);
|
||||
|
||||
adapter.poll();
|
||||
|
||||
assertEquals(3, finalAttempt.getAttemptCount());
|
||||
assertEquals("DLQ", finalAttempt.getStatus());
|
||||
assertEquals("processing failed", finalAttempt.getLastError());
|
||||
verify(repository, times(2)).save(finalAttempt);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1 @@
|
|||
mock-maker-subclass
|
||||
Loading…
Reference in New Issue