Repository: gchq/sleeper
Issue: Compaction test fails if a job starts before it's reported as created

When you run the compaction performance system test, sometimes a compaction job may be pulled off the queue by a compaction task before the status update is recorded in DynamoDB to report that the job has been created. If that happens, the job start time can be recorded as happening before the creation of the job.

There's a bug in the compaction reporting code where if a compaction job is started before it's created, it is not shown in the report.

The compaction performance test code uses the same logic to detect when a compaction job was created. If it can't see a compaction job it's created, it thinks there's nothing left to do and stops.

Changed files:
java/compaction/compaction-core/src/test/java/sleeper/compaction/job/CompactionJobStatusTest.java
java/compaction/compaction-core/src/main/java/sleeper/compaction/job/status/CompactionJobStatus.java
java/compaction/compaction-core/src/test/java/sleeper/compaction/job/CompactionJobStatusFromRecordsTest.java
java/core/src/main/java/sleeper/core/record/process/status/JobStatusUpdates.java

diff --git a/java/compaction/compaction-core/src/main/java/sleeper/compaction/job/status/CompactionJobStatus.java b/java/compaction/compaction-core/src/main/java/sleeper/compaction/job/status/CompactionJobStatus.java
index 4cf12ba81..dbb20eaae 100644
--- a/java/compaction/compaction-core/src/main/java/sleeper/compaction/job/status/CompactionJobStatus.java
+++ b/java/compaction/compaction-core/src/main/java/sleeper/compaction/job/status/CompactionJobStatus.java
@@ -18,7 +18,6 @@ package sleeper.compaction.job.status;
 import sleeper.core.record.process.status.JobStatusUpdates;
 import sleeper.core.record.process.status.ProcessRun;
 import sleeper.core.record.process.status.ProcessRuns;
-import sleeper.core.record.process.status.ProcessStatusUpdate;
 import sleeper.core.record.process.status.ProcessStatusUpdateRecord;
 
 import java.time.Instant;
@@ -59,15 +58,13 @@ public class CompactionJobStatus {
     }
 
     private static Optional<CompactionJobStatus> from(JobStatusUpdates updates) {
-        ProcessStatusUpdate firstUpdate = updates.getFirstRecord().getStatusUpdate();
-        if (!(firstUpdate instanceof CompactionJobCreatedStatus)) {
-            return Optional.empty();
-        }
-        return Optional.of(builder().jobId(updates.getJobId())
-                .createdStatus((CompactionJobCreatedStatus) firstUpdate)
-                .jobRuns(updates.getRuns())
-                .expiryDate(updates.getFirstRecord().getExpiryDate())
-                .build());
+        return updates.getFirstStatusUpdateOfType(CompactionJobCreatedStatus.class)
+                .map(createdStatus -> builder()
+                        .jobId(updates.getJobId())
+                        .createdStatus(createdStatus)
+                        .jobRuns(updates.getRuns())
+                        .expiryDate(updates.getFirstRecord().getExpiryDate())
+                        .build());
     }
 
     public Instant getCreateUpdateTime() {
diff --git a/java/compaction/compaction-core/src/test/java/sleeper/compaction/job/CompactionJobStatusFromRecordsTest.java b/java/compaction/compaction-core/src/test/java/sleeper/compaction/job/CompactionJobStatusFromRecordsTest.java
index 71f68fb78..fd2c7e3f9 100644
--- a/java/compaction/compaction-core/src/test/java/sleeper/compaction/job/CompactionJobStatusFromRecordsTest.java
+++ b/java/compaction/compaction-core/src/test/java/sleeper/compaction/job/CompactionJobStatusFromRecordsTest.java
@@ -36,10 +36,10 @@ import static sleeper.core.record.process.status.TestProcessStatusUpdateRecords.
 import static sleeper.core.record.process.status.TestProcessStatusUpdateRecords.forJob;
 import static sleeper.core.record.process.status.TestRunStatusUpdates.finishedStatus;
 
-public class CompactionJobStatusFromRecordsTest {
+class CompactionJobStatusFromRecordsTest {
 
     @Test
-    public void shouldBuildCompactionJobStatusFromIndividualUpdates() {
+    void shouldBuildCompactionJobStatusFromIndividualUpdates() {
         // Given
         CompactionJobCreatedStatus created1 = CompactionJobCreatedStatus.builder()
                 .updateTime(Instant.parse("2022-09-23T09:23:00.012Z"))
@@ -72,7 +72,7 @@ public class CompactionJobStatusFromRecordsTest {
     }
 
     @Test
-    public void shouldIgnoreJobWithNoCreatedUpdate() {
+    void shouldIgnoreJobWithNoCreatedUpdate() {
         // Given
         CompactionJobStartedStatus started = startedCompactionStatus(Instant.parse("2022-09-23T09:23:30.001Z"));
         ProcessFinishedStatus finished = finishedStatus(started, Duration.ofSeconds(30), 200L, 100L);
@@ -84,4 +84,26 @@ public class CompactionJobStatusFromRecordsTest {
         // Then
         assertThat(statuses).isEmpty();
     }
+
+    @Test
+    void shouldBuildJobStatusWhenCreatedUpdateStoredAfterStartedUpdate() {
+        // Given
+        CompactionJobCreatedStatus created = CompactionJobCreatedStatus.builder()
+                .updateTime(Instant.parse("2023-03-22T15:36:02Z"))
+                .partitionId("partition1").childPartitionIds(null)
+                .inputFilesCount(11)
+                .build();
+        CompactionJobStartedStatus started = startedCompactionStatus(Instant.parse("2023-03-22T15:36:01Z"));
+        ProcessFinishedStatus finished = finishedStatus(started, Duration.ofSeconds(30), 200L, 100L);
+
+        // When
+        List<CompactionJobStatus> statuses = jobStatusListFromUpdates(
+                forJob("test-job", created, started, finished));
+
+        // Then
+        assertThat(statuses).containsExactly(
+                CompactionJobStatus.builder().jobId("test-job").createdStatus(created)
+                        .singleJobRun(ProcessRun.finished(DEFAULT_TASK_ID, started, finished))
+                        .expiryDate(DEFAULT_EXPIRY).build());
+    }
 }
diff --git a/java/compaction/compaction-core/src/test/java/sleeper/compaction/job/CompactionJobStatusTest.java b/java/compaction/compaction-core/src/test/java/sleeper/compaction/job/CompactionJobStatusTest.java
index d28bee255..88fcdf730 100644
--- a/java/compaction/compaction-core/src/test/java/sleeper/compaction/job/CompactionJobStatusTest.java
+++ b/java/compaction/compaction-core/src/test/java/sleeper/compaction/job/CompactionJobStatusTest.java
@@ -32,12 +32,12 @@ import static sleeper.compaction.job.CompactionJobStatusTestData.jobCreated;
 import static sleeper.compaction.job.CompactionJobStatusTestData.startedCompactionRun;
 import static sleeper.core.record.process.status.TestProcessStatusUpdateRecords.DEFAULT_TASK_ID;
 
-public class CompactionJobStatusTest {
+class CompactionJobStatusTest {
 
     private final CompactionJobTestDataHelper dataHelper = new CompactionJobTestDataHelper();
 
     @Test
-    public void shouldBuildCompactionJobCreatedFromJob() {
+    void shouldBuildCompactionJobCreatedFromJob() {
         // Given
         Partition partition = dataHelper.singlePartition();
         CompactionJob job = dataHelper.singleFileCompaction(partition);
@@ -52,7 +52,7 @@ public class CompactionJobStatusTest {
     }
 
     @Test
-    public void shouldBuildSplittingCompactionJobCreatedFromJob() {
+    void shouldBuildSplittingCompactionJobCreatedFromJob() {
         // Given
         CompactionJob job = dataHelper.singleFileSplittingCompaction("root", "left", "right");
         Instant updateTime = Instant.parse("2022-09-22T13:33:12.001Z");
@@ -66,7 +66,7 @@ public class CompactionJobStatusTest {
     }
 
     @Test
-    public void shouldReportCompactionJobNotStarted() {
+    void shouldReportCompactionJobNotStarted() {
         // Given
         CompactionJob job = dataHelper.singleFileCompaction();
         Instant updateTime = Instant.parse("2022-09-22T13:33:12.001Z");
@@ -80,7 +80,7 @@ public class CompactionJobStatusTest {
     }
 
     @Test
-    public void shouldBuildCompactionJobStarted() {
+    void shouldBuildCompactionJobStarted() {
         // Given
         CompactionJob job = dataHelper.singleFileCompaction();
 
@@ -94,7 +94,7 @@ public class CompactionJobStatusTest {
     }
 
     @Test
-    public void shouldBuildCompactionJobFinished() {
+    void shouldBuildCompactionJobFinished() {
         // Given
         CompactionJob job = dataHelper.singleFileCompaction();
         Instant startTime = Instant.parse("2022-09-22T13:33:10.001Z");
diff --git a/java/core/src/main/java/sleeper/core/record/process/status/JobStatusUpdates.java b/java/core/src/main/java/sleeper/core/record/process/status/JobStatusUpdates.java
index 49ab61400..13ab5511d 100644
--- a/java/core/src/main/java/sleeper/core/record/process/status/JobStatusUpdates.java
+++ b/java/core/src/main/java/sleeper/core/record/process/status/JobStatusUpdates.java
@@ -17,6 +17,7 @@ package sleeper.core.record.process.status;
 
 import java.util.Comparator;
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -57,6 +58,16 @@ public class JobStatusUpdates {
         return recordsLatestFirst.get(0);
     }
 
+    public <T extends ProcessStatusUpdate> Optional<T> getFirstStatusUpdateOfType(Class<T> updateType) {
+        for (int i = recordsLatestFirst.size() - 1; i >= 0; i--) {
+            ProcessStatusUpdate update = recordsLatestFirst.get(i).getStatusUpdate();
+            if (updateType.isInstance(update)) {
+                return Optional.of(updateType.cast(update));
+            }
+        }
+        return Optional.empty();
+    }
+
     public ProcessRuns getRuns() {
         return runs;
     }