Repository: gchq/sleeper
Issue: DynamoDB connection closed before compaction task reported as finished

When a compaction task finishes, the DynamoDB client is shut down before it reports the task as finished. Sometimes it still manages to report it as finished, but often the connection is closed first and it fails.

Changed files:
java/compaction-status-store/src/test/java/sleeper/compaction/status/testutils/DynamoDBCompactionTaskStatusStoreTestBase.java
java/clients/src/test/java/sleeper/status/report/compactiontask/CompactionTaskStatusReportTest.java
java/compaction-status-store/src/test/java/sleeper/compaction/status/task/QueryCompactionTasksInProgressIT.java
java/compaction-core/src/main/java/sleeper/compaction/task/CompactionTaskStatus.java
java/compaction-job-execution/src/test/java/sleeper/compaction/jobexecution/CompactSortedFilesRunnerIT.java
java/compaction-core/src/test/java/sleeper/compaction/task/CompactionTaskStatusTest.java
java/compaction-status-store/src/test/java/sleeper/compaction/status/task/QueryCompactionTasksByPeriodIT.java
java/compaction-core/src/test/java/sleeper/compaction/task/CompactionTaskStatusInPeriodTest.java
java/compaction-job-execution/src/main/java/sleeper/compaction/jobexecution/CompactSortedFilesRunner.java

diff --git a/java/clients/src/test/java/sleeper/status/report/compactiontask/CompactionTaskStatusReportTest.java b/java/clients/src/test/java/sleeper/status/report/compactiontask/CompactionTaskStatusReportTest.java
index 67e29a91c..6d0101f93 100644
--- a/java/clients/src/test/java/sleeper/status/report/compactiontask/CompactionTaskStatusReportTest.java
+++ b/java/clients/src/test/java/sleeper/status/report/compactiontask/CompactionTaskStatusReportTest.java
@@ -44,7 +44,7 @@ public class CompactionTaskStatusReportTest {
     @Test
     public void shouldReportCompactionTaskUnfinished() throws Exception {
         // Given
-        CompactionTaskStatus task = CompactionTaskStatus
+        CompactionTaskStatus task = CompactionTaskStatus.builder()
                 .started(Instant.parse("2022-10-06T12:17:00.001Z"))
                 .taskId("A").build();
         when(store.getTasksInProgress()).thenReturn(Collections.singletonList(task));
@@ -59,10 +59,10 @@ public class CompactionTaskStatusReportTest {
     @Test
     public void shouldReportCompactionTaskUnfinishedAndFinished() throws Exception {
         // Given
-        CompactionTaskStatus unfinishedTask = CompactionTaskStatus
+        CompactionTaskStatus unfinishedTask = CompactionTaskStatus.builder()
                 .started(Instant.parse("2022-10-06T12:17:00.001Z"))
                 .taskId("unfinished-task").build();
-        CompactionTaskStatus finishedTask = CompactionTaskStatus
+        CompactionTaskStatus finishedTask = CompactionTaskStatus.builder()
                 .started(Instant.parse("2022-10-06T12:20:00.001Z"))
                 .taskId("finished-task")
                 .finished(CompactionTaskFinishedStatus.builder()
@@ -98,4 +98,8 @@ public class CompactionTaskStatusReportTest {
                 query).run();
         return output.toString();
     }
+
+    private static CompactionTaskStatus.Builder startedStatusBuilder(Instant startTime) {
+        return CompactionTaskStatus.builder().taskId("test-task-id").started(startTime);
+    }
 }
diff --git a/java/compaction-core/src/main/java/sleeper/compaction/task/CompactionTaskStatus.java b/java/compaction-core/src/main/java/sleeper/compaction/task/CompactionTaskStatus.java
index a08afcbf1..9e893b5b3 100644
--- a/java/compaction-core/src/main/java/sleeper/compaction/task/CompactionTaskStatus.java
+++ b/java/compaction-core/src/main/java/sleeper/compaction/task/CompactionTaskStatus.java
@@ -18,7 +18,6 @@ package sleeper.compaction.task;
 
 import java.time.Instant;
 import java.util.Objects;
-import java.util.UUID;
 
 public class CompactionTaskStatus {
     private final String taskId;
@@ -33,16 +32,6 @@ public class CompactionTaskStatus {
         expiryDate = builder.expiryDate;
     }
 
-    public static CompactionTaskStatus.Builder started(long startTime) {
-        return started(Instant.ofEpochMilli(startTime));
-    }
-
-    public static CompactionTaskStatus.Builder started(Instant startTime) {
-        return builder().taskId(UUID.randomUUID().toString())
-                .started(startTime);
-    }
-
-
     public static Builder builder() {
         return new Builder();
     }
diff --git a/java/compaction-core/src/test/java/sleeper/compaction/task/CompactionTaskStatusInPeriodTest.java b/java/compaction-core/src/test/java/sleeper/compaction/task/CompactionTaskStatusInPeriodTest.java
index 1b1eb9d13..80cc81275 100644
--- a/java/compaction-core/src/test/java/sleeper/compaction/task/CompactionTaskStatusInPeriodTest.java
+++ b/java/compaction-core/src/test/java/sleeper/compaction/task/CompactionTaskStatusInPeriodTest.java
@@ -31,8 +31,8 @@ public class CompactionTaskStatusInPeriodTest {
     @Test
     public void shouldBeInPeriodWithStartTimeOnly() {
         // Given
-        CompactionTaskStatus task = CompactionTaskStatus.started(
-                Instant.parse("2022-10-06T11:19:00.001Z")).build();
+        CompactionTaskStatus task = taskWithStartTime(
+                Instant.parse("2022-10-06T11:19:00.001Z"));
 
         // When / Then
         assertThat(task.isInPeriod(EPOCH_START, FAR_FUTURE)).isTrue();
@@ -42,7 +42,7 @@ public class CompactionTaskStatusInPeriodTest {
     public void shouldNotBeInPeriodWithStartTimeOnlyWhenEndIsStartTime() {
         // Given
         Instant startTime = Instant.parse("2022-10-06T11:19:00.001Z");
-        CompactionTaskStatus task = CompactionTaskStatus.started(startTime).build();
+        CompactionTaskStatus task = taskWithStartTime(startTime);
 
         // When / Then
         assertThat(task.isInPeriod(EPOCH_START, startTime)).isFalse();
@@ -52,7 +52,7 @@ public class CompactionTaskStatusInPeriodTest {
     public void shouldNotBeInPeriodWithStartTimeOnlyWhenStartIsStartTime() {
         // Given
         Instant startTime = Instant.parse("2022-10-06T11:19:00.001Z");
-        CompactionTaskStatus task = CompactionTaskStatus.started(startTime).build();
+        CompactionTaskStatus task = taskWithStartTime(startTime);
 
         // When / Then
         assertThat(task.isInPeriod(startTime, FAR_FUTURE)).isFalse();
@@ -113,8 +113,12 @@ public class CompactionTaskStatusInPeriodTest {
         assertThat(task.isInPeriod(startTime, FAR_FUTURE)).isTrue();
     }
 
+    private static CompactionTaskStatus taskWithStartTime(Instant startTime) {
+        return taskBuilder(startTime).build();
+    }
+
     private static CompactionTaskStatus taskWithStartAndFinishTime(Instant startTime, Instant finishTime) {
-        return CompactionTaskStatus.started(startTime)
+        return taskBuilder(startTime)
                 .finished(CompactionTaskFinishedStatus.builder()
                         .addJobSummary(new CompactionJobSummary(
                                 new CompactionJobRecordsProcessed(200, 100),
@@ -122,4 +126,8 @@ public class CompactionTaskStatusInPeriodTest {
                         )), finishTime)
                 .build();
     }
+
+    private static CompactionTaskStatus.Builder taskBuilder(Instant startTime) {
+        return CompactionTaskStatus.builder().taskId("test-task-id").started(startTime);
+    }
 }
diff --git a/java/compaction-core/src/test/java/sleeper/compaction/task/CompactionTaskStatusTest.java b/java/compaction-core/src/test/java/sleeper/compaction/task/CompactionTaskStatusTest.java
index a86c32a8f..5fae8349a 100644
--- a/java/compaction-core/src/test/java/sleeper/compaction/task/CompactionTaskStatusTest.java
+++ b/java/compaction-core/src/test/java/sleeper/compaction/task/CompactionTaskStatusTest.java
@@ -34,7 +34,7 @@ public class CompactionTaskStatusTest {
         Instant taskStartedTime = Instant.parse("2022-09-22T12:00:14.000Z");
 
         // When
-        CompactionTaskStatus status = CompactionTaskStatus.started(taskStartedTime.toEpochMilli()).build();
+        CompactionTaskStatus status = startedStatusBuilder(taskStartedTime).build();
 
         // Then
         assertThat(status).extracting("startedStatus.startTime")
@@ -48,7 +48,7 @@ public class CompactionTaskStatusTest {
         Instant jobFinishTime3 = Instant.parse("2022-09-22T16:00:14.000Z");
 
         // When
-        CompactionTaskStatus.Builder taskStatusBuilder = CompactionTaskStatus.started(taskStartedTime.toEpochMilli());
+        CompactionTaskStatus.Builder taskStatusBuilder = startedStatusBuilder(taskStartedTime);
         CompactionTaskFinishedStatus.Builder taskFinishedBuilder = CompactionTaskFinishedStatus.builder();
         createJobSummaries().forEach(taskFinishedBuilder::addJobSummary);
         taskStatusBuilder.finished(taskFinishedBuilder, jobFinishTime3.toEpochMilli());
@@ -67,7 +67,7 @@ public class CompactionTaskStatusTest {
         Instant jobFinishTime3 = Instant.parse("2022-09-22T16:00:14.000Z");
 
         // When
-        CompactionTaskStatus.Builder taskStatusBuilder = CompactionTaskStatus.started(taskStartedTime.toEpochMilli());
+        CompactionTaskStatus.Builder taskStatusBuilder = startedStatusBuilder(taskStartedTime);
         CompactionTaskFinishedStatus.Builder taskFinishedBuilder = CompactionTaskFinishedStatus.builder();
         createJobSummaries().forEach(taskFinishedBuilder::addJobSummary);
         taskStatusBuilder.finished(taskFinishedBuilder, jobFinishTime3.toEpochMilli());
@@ -79,7 +79,11 @@ public class CompactionTaskStatusTest {
                 .containsExactly(taskStartedTime, jobFinishTime3, 3, 14400.0, 14400L, 7200L, 1.0, 0.5);
     }
 
-    private List<CompactionJobSummary> createJobSummaries() {
+    private static CompactionTaskStatus.Builder startedStatusBuilder(Instant startTime) {
+        return CompactionTaskStatus.builder().taskId("test-task-id").started(startTime);
+    }
+
+    private static List<CompactionJobSummary> createJobSummaries() {
         Instant jobStartedUpdateTime1 = Instant.parse("2022-09-22T14:00:04.000Z");
         Instant jobFinishTime1 = Instant.parse("2022-09-22T14:00:14.000Z");
 
diff --git a/java/compaction-job-execution/src/main/java/sleeper/compaction/jobexecution/CompactSortedFilesRunner.java b/java/compaction-job-execution/src/main/java/sleeper/compaction/jobexecution/CompactSortedFilesRunner.java
index d4157e81b..d66751329 100644
--- a/java/compaction-job-execution/src/main/java/sleeper/compaction/jobexecution/CompactSortedFilesRunner.java
+++ b/java/compaction-job-execution/src/main/java/sleeper/compaction/jobexecution/CompactSortedFilesRunner.java
@@ -52,6 +52,8 @@ import sleeper.statestore.StateStoreProvider;
 import sleeper.utils.HadoopConfigurationProvider;
 
 import java.io.IOException;
+import java.time.Instant;
+import java.util.UUID;
 
 import static sleeper.configuration.properties.SystemDefinedInstanceProperty.COMPACTION_JOB_QUEUE_URL;
 import static sleeper.configuration.properties.SystemDefinedInstanceProperty.SPLITTING_COMPACTION_JOB_QUEUE_URL;
@@ -77,8 +79,8 @@ public class CompactSortedFilesRunner {
     private final TablePropertiesProvider tablePropertiesProvider;
     private final StateStoreProvider stateStoreProvider;
     private final CompactionJobStatusStore jobStatusStore;
+    private final CompactionTaskStatusStore taskStatusStore;
     private final String taskId;
-    private final CompactionTaskFinishedStatus.Builder taskFinishedBuilder;
     private final CompactionJobSerDe compactionJobSerDe;
     private final String sqsJobQueueUrl;
     private final AmazonSQS sqsClient;
@@ -93,7 +95,7 @@ public class CompactSortedFilesRunner {
             TablePropertiesProvider tablePropertiesProvider,
             StateStoreProvider stateStoreProvider,
             CompactionJobStatusStore jobStatusStore,
-            CompactionTaskFinishedStatus.Builder taskFinishedBuilder,
+            CompactionTaskStatusStore taskStatusStore,
             String taskId,
             String sqsJobQueueUrl,
             AmazonSQS sqsClient,
@@ -104,7 +106,7 @@ public class CompactSortedFilesRunner {
         this.tablePropertiesProvider = tablePropertiesProvider;
         this.stateStoreProvider = stateStoreProvider;
         this.jobStatusStore = jobStatusStore;
-        this.taskFinishedBuilder = taskFinishedBuilder;
+        this.taskStatusStore = taskStatusStore;
         this.taskId = taskId;
         this.compactionJobSerDe = new CompactionJobSerDe(tablePropertiesProvider);
         this.sqsJobQueueUrl = sqsJobQueueUrl;
@@ -121,14 +123,21 @@ public class CompactSortedFilesRunner {
             TablePropertiesProvider tablePropertiesProvider,
             StateStoreProvider stateStoreProvider,
             CompactionJobStatusStore jobStatusStore,
-            CompactionTaskFinishedStatus.Builder taskFinishedBuilder,
+            CompactionTaskStatusStore taskStatusStore,
             String taskId,
             String sqsJobQueueUrl,
             AmazonSQS sqsClient) {
-        this(instanceProperties, objectFactory, tablePropertiesProvider, stateStoreProvider, jobStatusStore, taskFinishedBuilder, taskId, sqsJobQueueUrl, sqsClient, 3, 20);
+        this(instanceProperties, objectFactory, tablePropertiesProvider, stateStoreProvider, jobStatusStore, taskStatusStore, taskId, sqsJobQueueUrl, sqsClient, 3, 20);
     }
 
     public void run() throws InterruptedException, IOException, ActionException {
+
+        Instant startTime = Instant.now();
+        CompactionTaskStatus.Builder taskStatusBuilder = CompactionTaskStatus
+                .builder().taskId(taskId).started(startTime);
+        LOGGER.info("Starting task {}", taskId);
+        taskStatusStore.taskStarted(taskStatusBuilder.build());
+        CompactionTaskFinishedStatus.Builder taskFinishedBuilder = CompactionTaskFinishedStatus.builder();
         long totalNumberOfMessagesProcessed = 0L;
         int numConsecutiveTimesNoMessages = 0;
         while (numConsecutiveTimesNoMessages < maxMessageRetrieveAttempts) {
@@ -157,7 +166,14 @@ public class CompactSortedFilesRunner {
         }
         LOGGER.info("Returning from run() method in CompactSortedFilesRunner as no messages received in {} seconds",
                 (numConsecutiveTimesNoMessages * 30));
-        LOGGER.info("Total number of messages processed = " + totalNumberOfMessagesProcessed);
+        LOGGER.info("Total number of messages processed = {}", totalNumberOfMessagesProcessed);
+
+        Instant finishTime = Instant.now();
+        double runTimeInSeconds = (finishTime.toEpochMilli() - startTime.toEpochMilli()) / 1000.0;
+        LOGGER.info("CompactSortedFilesRunner total run time = {}", runTimeInSeconds);
+
+        CompactionTaskStatus taskFinished = taskStatusBuilder.finished(taskFinishedBuilder, finishTime).build();
+        taskStatusStore.taskFinished(taskFinished);
     }
 
     private CompactionJobSummary compact(CompactionJob compactionJob, Message message) throws IOException, IteratorException, ActionException {
@@ -198,7 +214,6 @@ public class CompactSortedFilesRunner {
             System.exit(1);
         }
 
-        long startTime = System.currentTimeMillis();
         AmazonDynamoDB dynamoDBClient = AmazonDynamoDBClientBuilder.defaultClient();
         AmazonSQS sqsClient = AmazonSQSClientBuilder.defaultClient();
         AmazonS3 s3Client = AmazonS3ClientBuilder.defaultClient();
@@ -211,10 +226,6 @@ public class CompactSortedFilesRunner {
         StateStoreProvider stateStoreProvider = new StateStoreProvider(dynamoDBClient, instanceProperties, HadoopConfigurationProvider.getConfigurationForECS(instanceProperties));
         CompactionJobStatusStore jobStatusStore = DynamoDBCompactionJobStatusStore.from(dynamoDBClient, instanceProperties);
         CompactionTaskStatusStore taskStatusStore = DynamoDBCompactionTaskStatusStore.from(dynamoDBClient, instanceProperties);
-        CompactionTaskFinishedStatus.Builder taskFinishedBuilder = CompactionTaskFinishedStatus.builder();
-
-        CompactionTaskStatus.Builder taskStatusBuilder = CompactionTaskStatus.started(startTime);
-        taskStatusStore.taskStarted(taskStatusBuilder.build());
 
         String sqsJobQueueUrl;
         String type = args[1];
@@ -232,8 +243,8 @@ public class CompactSortedFilesRunner {
                 tablePropertiesProvider,
                 stateStoreProvider,
                 jobStatusStore,
-                taskFinishedBuilder,
-                taskStatusBuilder.getTaskId(),
+                taskStatusStore,
+                UUID.randomUUID().toString(),
                 sqsJobQueueUrl,
                 sqsClient);
         runner.run();
@@ -244,11 +255,5 @@ public class CompactSortedFilesRunner {
         LOGGER.info("Shut down dynamoDBClient");
         s3Client.shutdown();
         LOGGER.info("Shut down s3Client");
-        long finishTime = System.currentTimeMillis();
-        double runTimeInSeconds = (finishTime - startTime) / 1000.0;
-        LOGGER.info("CompactSortedFilesRunner total run time = " + runTimeInSeconds);
-
-        CompactionTaskStatus taskFinished = taskStatusBuilder.finished(taskFinishedBuilder, finishTime).build();
-        taskStatusStore.taskFinished(taskFinished);
     }
 }
diff --git a/java/compaction-job-execution/src/test/java/sleeper/compaction/jobexecution/CompactSortedFilesRunnerIT.java b/java/compaction-job-execution/src/test/java/sleeper/compaction/jobexecution/CompactSortedFilesRunnerIT.java
index efedc916c..65af6db17 100644
--- a/java/compaction-job-execution/src/test/java/sleeper/compaction/jobexecution/CompactSortedFilesRunnerIT.java
+++ b/java/compaction-job-execution/src/test/java/sleeper/compaction/jobexecution/CompactSortedFilesRunnerIT.java
@@ -36,7 +36,9 @@ import sleeper.compaction.job.CompactionJobSerDe;
 import sleeper.compaction.job.CompactionJobStatusStore;
 import sleeper.compaction.status.job.DynamoDBCompactionJobStatusStore;
 import sleeper.compaction.status.job.DynamoDBCompactionJobStatusStoreCreator;
-import sleeper.compaction.task.CompactionTaskFinishedStatus;
+import sleeper.compaction.status.task.DynamoDBCompactionTaskStatusStore;
+import sleeper.compaction.status.task.DynamoDBCompactionTaskStatusStoreCreator;
+import sleeper.compaction.task.CompactionTaskStatusStore;
 import sleeper.configuration.jars.ObjectFactory;
 import sleeper.configuration.jars.ObjectFactoryException;
 import sleeper.configuration.properties.InstanceProperties;
@@ -156,6 +158,8 @@ public class CompactSortedFilesRunnerIT {
         stateStore.initialise();
         DynamoDBCompactionJobStatusStoreCreator.create(instanceProperties, dynamoDB);
         CompactionJobStatusStore jobStatusStore = DynamoDBCompactionJobStatusStore.from(dynamoDB, instanceProperties);
+        DynamoDBCompactionTaskStatusStoreCreator.create(instanceProperties, dynamoDB);
+        CompactionTaskStatusStore taskStatusStore = DynamoDBCompactionTaskStatusStore.from(dynamoDB, instanceProperties);
         //  - Create four files of sorted data
         String folderName = folder.newFolder().getAbsolutePath();
         String file1 = folderName + "/file1.parquet";
@@ -268,7 +272,7 @@ public class CompactSortedFilesRunnerIT {
         // When
         CompactSortedFilesRunner runner = new CompactSortedFilesRunner(
                 instanceProperties, ObjectFactory.noUserJars(),
-                tablePropertiesProvider, stateStoreProvider, jobStatusStore, CompactionTaskFinishedStatus.builder(),
+                tablePropertiesProvider, stateStoreProvider, jobStatusStore, taskStatusStore,
                 "task-id", instanceProperties.get(COMPACTION_JOB_QUEUE_URL), sqsClient,
                 1, 5);
         runner.run();
diff --git a/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/QueryCompactionTasksByPeriodIT.java b/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/QueryCompactionTasksByPeriodIT.java
index aa3fc9bbf..ddd8cc093 100644
--- a/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/QueryCompactionTasksByPeriodIT.java
+++ b/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/QueryCompactionTasksByPeriodIT.java
@@ -24,6 +24,7 @@ import sleeper.compaction.task.CompactionTaskStatus;
 
 import java.time.Instant;
 import java.time.Period;
+import java.util.UUID;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -90,7 +91,7 @@ public class QueryCompactionTasksByPeriodIT extends DynamoDBCompactionTaskStatus
     }
 
     private CompactionTaskStatus taskWithStartAndFinishTime(Instant startTime, Instant finishTime) {
-        return CompactionTaskStatus.started(startTime)
+        return CompactionTaskStatus.builder().taskId(UUID.randomUUID().toString()).started(startTime)
                 .finished(CompactionTaskFinishedStatus.builder()
                         .addJobSummary(new CompactionJobSummary(
                                 new CompactionJobRecordsProcessed(200, 100),
diff --git a/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/QueryCompactionTasksInProgressIT.java b/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/QueryCompactionTasksInProgressIT.java
index b01aa5fd1..1ec4417b2 100644
--- a/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/QueryCompactionTasksInProgressIT.java
+++ b/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/QueryCompactionTasksInProgressIT.java
@@ -55,10 +55,10 @@ public class QueryCompactionTasksInProgressIT extends DynamoDBCompactionTaskStat
     @Test
     public void shouldSortByStartTimeMostRecentFirst() {
         // Given
-        CompactionTaskStatus task1 = CompactionTaskStatus.started(
-                Instant.parse("2022-10-06T11:19:00.001Z")).build();
-        CompactionTaskStatus task2 = CompactionTaskStatus.started(
-                Instant.parse("2022-10-06T11:19:10.001Z")).build();
+        CompactionTaskStatus task1 = startedTaskWithDefaultsBuilder()
+                .started(Instant.parse("2022-10-06T11:19:00.001Z")).build();
+        CompactionTaskStatus task2 = startedTaskWithDefaultsBuilder()
+                .started(Instant.parse("2022-10-06T11:19:10.001Z")).build();
 
         // When
         store.taskStarted(task1);
diff --git a/java/compaction-status-store/src/test/java/sleeper/compaction/status/testutils/DynamoDBCompactionTaskStatusStoreTestBase.java b/java/compaction-status-store/src/test/java/sleeper/compaction/status/testutils/DynamoDBCompactionTaskStatusStoreTestBase.java
index bad25e1fb..3d3cf710f 100644
--- a/java/compaction-status-store/src/test/java/sleeper/compaction/status/testutils/DynamoDBCompactionTaskStatusStoreTestBase.java
+++ b/java/compaction-status-store/src/test/java/sleeper/compaction/status/testutils/DynamoDBCompactionTaskStatusStoreTestBase.java
@@ -28,6 +28,7 @@ import sleeper.compaction.task.CompactionTaskStatusStore;
 import sleeper.configuration.properties.InstanceProperties;
 
 import java.time.Instant;
+import java.util.UUID;
 
 import static sleeper.compaction.status.task.DynamoDBCompactionTaskStatusStore.taskStatusTableName;
 import static sleeper.compaction.status.testutils.CompactionStatusStoreTestUtils.createInstanceProperties;
@@ -64,7 +65,7 @@ public class DynamoDBCompactionTaskStatusStoreTestBase extends DynamoDBTestBase
     }
 
     protected static CompactionTaskStatus.Builder startedTaskWithDefaultsBuilder() {
-        return CompactionTaskStatus.started(defaultStartTime().toEpochMilli());
+        return CompactionTaskStatus.builder().taskId(UUID.randomUUID().toString()).started(defaultStartTime());
     }
 
     protected static CompactionTaskStatus finishedTaskWithDefaults() {