If sleeper.compaction.status.store.enabled is set to false, compaction crashes on reporting task status
When the status store is disabled, no DynamoDB table is created for compaction tasks, but the status store DynamoDB implementation is still used. This means all compaction tasks will crash at startup as they try to report that they've started.
diff --git a/java/cdk/src/main/java/sleeper/cdk/Utils.java b/java/cdk/src/main/java/sleeper/cdk/Utils.java
index 371c11092..c752ef7f5 100644
--- a/java/cdk/src/main/java/sleeper/cdk/Utils.java
+++ b/java/cdk/src/main/java/sleeper/cdk/Utils.java
@@ -193,7 +193,7 @@ public class Utils {
     }
 
     public static RemovalPolicy removalPolicy(InstanceProperties properties) {
-        if (Boolean.TRUE.equals(properties.getBoolean(RETAIN_INFRA_AFTER_DESTROY))) {
+        if (properties.getBoolean(RETAIN_INFRA_AFTER_DESTROY)) {
             return RemovalPolicy.RETAIN;
         } else {
             return RemovalPolicy.DESTROY;
diff --git a/java/compaction-core/src/main/java/sleeper/compaction/job/CompactionJobStatusStore.java b/java/compaction-core/src/main/java/sleeper/compaction/job/CompactionJobStatusStore.java
index 9700df1ff..9725418bf 100644
--- a/java/compaction-core/src/main/java/sleeper/compaction/job/CompactionJobStatusStore.java
+++ b/java/compaction-core/src/main/java/sleeper/compaction/job/CompactionJobStatusStore.java
@@ -23,10 +23,8 @@ import java.util.Optional;
 
 public interface CompactionJobStatusStore {
 
-    static CompactionJobStatusStore none() {
-        return new CompactionJobStatusStore() {
-        };
-    }
+    CompactionJobStatusStore NONE = new CompactionJobStatusStore() {
+    };
 
     default void jobCreated(CompactionJob job) {
     }
diff --git a/java/compaction-core/src/main/java/sleeper/compaction/task/CompactionTaskStatusStore.java b/java/compaction-core/src/main/java/sleeper/compaction/task/CompactionTaskStatusStore.java
index bdff9452e..267e17577 100644
--- a/java/compaction-core/src/main/java/sleeper/compaction/task/CompactionTaskStatusStore.java
+++ b/java/compaction-core/src/main/java/sleeper/compaction/task/CompactionTaskStatusStore.java
@@ -16,17 +16,13 @@
 
 package sleeper.compaction.task;
 
-import sleeper.compaction.job.CompactionJobStatusStore;
-
 import java.time.Instant;
 import java.util.List;
 
 public interface CompactionTaskStatusStore {
 
-    static CompactionJobStatusStore none() {
-        return new CompactionJobStatusStore() {
-        };
-    }
+    CompactionTaskStatusStore NONE = new CompactionTaskStatusStore() {
+    };
 
     default void taskStarted(CompactionTaskStatus taskStatus) {
     }
diff --git a/java/compaction-job-execution/src/test/java/sleeper/compaction/jobexecution/testutils/CompactSortedFilesTestUtils.java b/java/compaction-job-execution/src/test/java/sleeper/compaction/jobexecution/testutils/CompactSortedFilesTestUtils.java
index a8b9cc208..e28f725af 100644
--- a/java/compaction-job-execution/src/test/java/sleeper/compaction/jobexecution/testutils/CompactSortedFilesTestUtils.java
+++ b/java/compaction-job-execution/src/test/java/sleeper/compaction/jobexecution/testutils/CompactSortedFilesTestUtils.java
@@ -93,7 +93,7 @@ public class CompactSortedFilesTestUtils {
 
     public static CompactSortedFiles createCompactSortedFiles(
             Schema schema, CompactionJob compactionJob, StateStore stateStore, String taskId) {
-        return createCompactSortedFiles(schema, compactionJob, stateStore, CompactionJobStatusStore.none(), taskId);
+        return createCompactSortedFiles(schema, compactionJob, stateStore, CompactionJobStatusStore.NONE, taskId);
     }
 
     public static CompactSortedFiles createCompactSortedFiles(
diff --git a/java/compaction-status-store/src/main/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStore.java b/java/compaction-status-store/src/main/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStore.java
index dd4560434..4e3b0bc2a 100644
--- a/java/compaction-status-store/src/main/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStore.java
+++ b/java/compaction-status-store/src/main/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStore.java
@@ -64,10 +64,10 @@ public class DynamoDBCompactionJobStatusStore implements CompactionJobStatusStor
     }
 
     public static CompactionJobStatusStore from(AmazonDynamoDB dynamoDB, InstanceProperties properties) {
-        if (Boolean.TRUE.equals(properties.getBoolean(COMPACTION_STATUS_STORE_ENABLED))) {
+        if (properties.getBoolean(COMPACTION_STATUS_STORE_ENABLED)) {
             return new DynamoDBCompactionJobStatusStore(dynamoDB, properties);
         } else {
-            return CompactionJobStatusStore.none();
+            return CompactionJobStatusStore.NONE;
         }
     }
 
diff --git a/java/compaction-status-store/src/main/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStoreCreator.java b/java/compaction-status-store/src/main/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStoreCreator.java
index 082d84355..2607e5510 100644
--- a/java/compaction-status-store/src/main/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStoreCreator.java
+++ b/java/compaction-status-store/src/main/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStoreCreator.java
@@ -28,6 +28,7 @@ import static sleeper.compaction.status.job.DynamoDBCompactionJobStatusFormat.EX
 import static sleeper.compaction.status.job.DynamoDBCompactionJobStatusFormat.JOB_ID;
 import static sleeper.compaction.status.job.DynamoDBCompactionJobStatusFormat.UPDATE_TIME;
 import static sleeper.compaction.status.job.DynamoDBCompactionJobStatusStore.jobStatusTableName;
+import static sleeper.configuration.properties.UserDefinedInstanceProperty.COMPACTION_STATUS_STORE_ENABLED;
 import static sleeper.configuration.properties.UserDefinedInstanceProperty.ID;
 import static sleeper.dynamodb.tools.DynamoDBUtils.configureTimeToLive;
 import static sleeper.dynamodb.tools.DynamoDBUtils.initialiseTable;
@@ -38,6 +39,9 @@ public class DynamoDBCompactionJobStatusStoreCreator {
     }
 
     public static void create(InstanceProperties properties, AmazonDynamoDB dynamoDB) {
+        if (!properties.getBoolean(COMPACTION_STATUS_STORE_ENABLED)) {
+            return;
+        }
         String tableName = jobStatusTableName(properties.get(ID));
         initialiseTable(dynamoDB, tableName,
                 Arrays.asList(
@@ -48,4 +52,11 @@ public class DynamoDBCompactionJobStatusStoreCreator {
                         new KeySchemaElement(UPDATE_TIME, KeyType.RANGE)));
         configureTimeToLive(dynamoDB, tableName, EXPIRY_DATE);
     }
+
+    public static void tearDown(InstanceProperties properties, AmazonDynamoDB dynamoDBClient) {
+        if (!properties.getBoolean(COMPACTION_STATUS_STORE_ENABLED)) {
+            return;
+        }
+        dynamoDBClient.deleteTable(jobStatusTableName(properties.get(ID)));
+    }
 }
diff --git a/java/compaction-status-store/src/main/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStore.java b/java/compaction-status-store/src/main/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStore.java
index 7b8e8f79c..1d0a51327 100644
--- a/java/compaction-status-store/src/main/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStore.java
+++ b/java/compaction-status-store/src/main/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStore.java
@@ -40,6 +40,7 @@ import java.util.Map;
 import java.util.stream.Collectors;
 
 import static sleeper.compaction.status.task.DynamoDBCompactionTaskStatusFormat.TASK_ID;
+import static sleeper.configuration.properties.UserDefinedInstanceProperty.COMPACTION_STATUS_STORE_ENABLED;
 import static sleeper.configuration.properties.UserDefinedInstanceProperty.ID;
 import static sleeper.dynamodb.tools.DynamoDBAttributes.createStringAttribute;
 import static sleeper.dynamodb.tools.DynamoDBUtils.instanceTableName;
@@ -120,8 +121,12 @@ public class DynamoDBCompactionTaskStatusStore implements CompactionTaskStatusSt
         return dynamoDB.putItem(putItemRequest);
     }
 
-    public static DynamoDBCompactionTaskStatusStore from(AmazonDynamoDB dynamoDB, InstanceProperties properties) {
-        return new DynamoDBCompactionTaskStatusStore(dynamoDB, properties);
+    public static CompactionTaskStatusStore from(AmazonDynamoDB dynamoDB, InstanceProperties properties) {
+        if (properties.getBoolean(COMPACTION_STATUS_STORE_ENABLED)) {
+            return new DynamoDBCompactionTaskStatusStore(dynamoDB, properties);
+        } else {
+            return CompactionTaskStatusStore.NONE;
+        }
     }
 
     public static String taskStatusTableName(String instanceId) {
diff --git a/java/compaction-status-store/src/main/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStoreCreator.java b/java/compaction-status-store/src/main/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStoreCreator.java
index e78edec5f..844e8f154 100644
--- a/java/compaction-status-store/src/main/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStoreCreator.java
+++ b/java/compaction-status-store/src/main/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStoreCreator.java
@@ -28,6 +28,7 @@ import static sleeper.compaction.status.task.DynamoDBCompactionTaskStatusFormat.
 import static sleeper.compaction.status.task.DynamoDBCompactionTaskStatusFormat.TASK_ID;
 import static sleeper.compaction.status.task.DynamoDBCompactionTaskStatusFormat.UPDATE_TIME;
 import static sleeper.compaction.status.task.DynamoDBCompactionTaskStatusStore.taskStatusTableName;
+import static sleeper.configuration.properties.UserDefinedInstanceProperty.COMPACTION_STATUS_STORE_ENABLED;
 import static sleeper.configuration.properties.UserDefinedInstanceProperty.ID;
 import static sleeper.dynamodb.tools.DynamoDBUtils.configureTimeToLive;
 import static sleeper.dynamodb.tools.DynamoDBUtils.initialiseTable;
@@ -38,6 +39,9 @@ public class DynamoDBCompactionTaskStatusStoreCreator {
     }
 
     public static void create(InstanceProperties properties, AmazonDynamoDB dynamoDB) {
+        if (!properties.getBoolean(COMPACTION_STATUS_STORE_ENABLED)) {
+            return;
+        }
         String tableName = taskStatusTableName(properties.get(ID));
         initialiseTable(dynamoDB, tableName,
                 Arrays.asList(
@@ -48,4 +52,11 @@ public class DynamoDBCompactionTaskStatusStoreCreator {
                         new KeySchemaElement(UPDATE_TIME, KeyType.RANGE)));
         configureTimeToLive(dynamoDB, tableName, EXPIRY_DATE);
     }
+
+    public static void tearDown(InstanceProperties properties, AmazonDynamoDB dynamoDBClient) {
+        if (!properties.getBoolean(COMPACTION_STATUS_STORE_ENABLED)) {
+            return;
+        }
+        dynamoDBClient.deleteTable(taskStatusTableName(properties.get(ID)));
+    }
 }
diff --git a/java/compaction-status-store/src/test/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStoreCreatorIT.java b/java/compaction-status-store/src/test/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStoreCreatorIT.java
index 13003b38c..2bc06d252 100644
--- a/java/compaction-status-store/src/test/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStoreCreatorIT.java
+++ b/java/compaction-status-store/src/test/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStoreCreatorIT.java
@@ -16,14 +16,18 @@
 package sleeper.compaction.status.job;
 
 import com.amazonaws.services.dynamodbv2.model.DescribeTableResult;
+import com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException;
 import org.junit.After;
 import org.junit.Test;
+import sleeper.compaction.job.CompactionJobStatusStore;
 import sleeper.configuration.properties.InstanceProperties;
 import sleeper.dynamodb.tools.DynamoDBTestBase;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static sleeper.compaction.status.job.DynamoDBCompactionJobStatusStore.jobStatusTableName;
 import static sleeper.compaction.status.testutils.CompactionStatusStoreTestUtils.createInstanceProperties;
+import static sleeper.configuration.properties.UserDefinedInstanceProperty.COMPACTION_STATUS_STORE_ENABLED;
 import static sleeper.configuration.properties.UserDefinedInstanceProperty.ID;
 
 public class DynamoDBCompactionJobStatusStoreCreatorIT extends DynamoDBTestBase {
@@ -35,14 +39,34 @@ public class DynamoDBCompactionJobStatusStoreCreatorIT extends DynamoDBTestBase
     public void shouldCreateStore() {
         // When
         DynamoDBCompactionJobStatusStoreCreator.create(instanceProperties, dynamoDBClient);
+        CompactionJobStatusStore store = DynamoDBCompactionJobStatusStore.from(dynamoDBClient, instanceProperties);
 
         // Then
         assertThat(dynamoDBClient.describeTable(tableName))
                 .extracting(DescribeTableResult::getTable).isNotNull();
+        assertThat(store).isInstanceOf(DynamoDBCompactionJobStatusStore.class);
+    }
+
+    @Test
+    public void shouldNotCreateStoreIfDisabled() {
+        // Given
+        instanceProperties.set(COMPACTION_STATUS_STORE_ENABLED, "false");
+
+        // When
+        DynamoDBCompactionJobStatusStoreCreator.create(instanceProperties, dynamoDBClient);
+        CompactionJobStatusStore store = DynamoDBCompactionJobStatusStore.from(dynamoDBClient, instanceProperties);
+
+        // Then
+        assertThatThrownBy(() -> dynamoDBClient.describeTable(tableName))
+                .isInstanceOf(ResourceNotFoundException.class);
+        assertThat(store).isSameAs(CompactionJobStatusStore.NONE);
+        assertThatThrownBy(() -> store.getAllJobs("some-table")).isInstanceOf(UnsupportedOperationException.class);
+        assertThatThrownBy(() -> store.getUnfinishedJobs("some-table")).isInstanceOf(UnsupportedOperationException.class);
+        assertThatThrownBy(() -> store.getJob("some-job")).isInstanceOf(UnsupportedOperationException.class);
     }
 
     @After
     public void tearDown() {
-        dynamoDBClient.deleteTable(tableName);
+        DynamoDBCompactionJobStatusStoreCreator.tearDown(instanceProperties, dynamoDBClient);
     }
 }
diff --git a/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStoreCreatorIT.java b/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStoreCreatorIT.java
index c564fb02b..f7a901898 100644
--- a/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStoreCreatorIT.java
+++ b/java/compaction-status-store/src/test/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStoreCreatorIT.java
@@ -16,14 +16,18 @@
 package sleeper.compaction.status.task;
 
 import com.amazonaws.services.dynamodbv2.model.DescribeTableResult;
+import com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException;
 import org.junit.After;
 import org.junit.Test;
+import sleeper.compaction.task.CompactionTaskStatusStore;
 import sleeper.configuration.properties.InstanceProperties;
 import sleeper.dynamodb.tools.DynamoDBTestBase;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static sleeper.compaction.status.task.DynamoDBCompactionTaskStatusStore.taskStatusTableName;
 import static sleeper.compaction.status.testutils.CompactionStatusStoreTestUtils.createInstanceProperties;
+import static sleeper.configuration.properties.UserDefinedInstanceProperty.COMPACTION_STATUS_STORE_ENABLED;
 import static sleeper.configuration.properties.UserDefinedInstanceProperty.ID;
 
 public class DynamoDBCompactionTaskStatusStoreCreatorIT extends DynamoDBTestBase {
@@ -35,14 +39,34 @@ public class DynamoDBCompactionTaskStatusStoreCreatorIT extends DynamoDBTestBase
     public void shouldCreateStore() {
         // When
         DynamoDBCompactionTaskStatusStoreCreator.create(instanceProperties, dynamoDBClient);
+        CompactionTaskStatusStore store = DynamoDBCompactionTaskStatusStore.from(dynamoDBClient, instanceProperties);
 
         // Then
         assertThat(dynamoDBClient.describeTable(tableName))
                 .extracting(DescribeTableResult::getTable).isNotNull();
+        assertThat(store).isInstanceOf(DynamoDBCompactionTaskStatusStore.class);
+    }
+
+    @Test
+    public void shouldNotCreateStoreIfDisabled() {
+        // Given
+        instanceProperties.set(COMPACTION_STATUS_STORE_ENABLED, "false");
+
+        // When
+        DynamoDBCompactionTaskStatusStoreCreator.create(instanceProperties, dynamoDBClient);
+        CompactionTaskStatusStore store = DynamoDBCompactionTaskStatusStore.from(dynamoDBClient, instanceProperties);
+
+        // Then
+        assertThatThrownBy(() -> dynamoDBClient.describeTable(tableName))
+                .isInstanceOf(ResourceNotFoundException.class);
+        assertThat(store).isSameAs(CompactionTaskStatusStore.NONE);
+        assertThatThrownBy(store::getAllTasks).isInstanceOf(UnsupportedOperationException.class);
+        assertThatThrownBy(store::getTasksInProgress).isInstanceOf(UnsupportedOperationException.class);
+        assertThatThrownBy(() -> store.getTask("some-task")).isInstanceOf(UnsupportedOperationException.class);
     }
 
     @After
     public void tearDown() {
-        dynamoDBClient.deleteTable(tableName);
+        DynamoDBCompactionTaskStatusStoreCreator.tearDown(instanceProperties, dynamoDBClient);
     }
 }
diff --git a/java/configuration/src/main/java/sleeper/configuration/properties/SleeperProperties.java b/java/configuration/src/main/java/sleeper/configuration/properties/SleeperProperties.java
index 60aa6cad7..933e6080a 100644
--- a/java/configuration/src/main/java/sleeper/configuration/properties/SleeperProperties.java
+++ b/java/configuration/src/main/java/sleeper/configuration/properties/SleeperProperties.java
@@ -65,7 +65,7 @@ public abstract class SleeperProperties<T extends SleeperProperty> {
         return properties.getProperty(property.getPropertyName(), property.getDefaultValue());
     }
 
-    public Boolean getBoolean(T property) {
+    public boolean getBoolean(T property) {
         return Boolean.parseBoolean(get(property));
     }
 
diff --git a/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/job/status/DynamoDBIngestJobStatusStore.java b/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/job/status/DynamoDBIngestJobStatusStore.java
index 72cdb33fa..79b35dc4d 100644
--- a/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/job/status/DynamoDBIngestJobStatusStore.java
+++ b/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/job/status/DynamoDBIngestJobStatusStore.java
@@ -37,7 +37,7 @@ public class DynamoDBIngestJobStatusStore implements IngestJobStatusStore {
     }
 
     public static IngestJobStatusStore from(AmazonDynamoDB dynamoDB, InstanceProperties properties) {
-        if (Boolean.TRUE.equals(properties.getBoolean(INGEST_STATUS_STORE_ENABLED))) {
+        if (properties.getBoolean(INGEST_STATUS_STORE_ENABLED)) {
             return new DynamoDBIngestJobStatusStore(dynamoDB, properties);
         } else {
             return IngestJobStatusStore.none();
['java/cdk/src/main/java/sleeper/cdk/Utils.java', 'java/compaction-core/src/main/java/sleeper/compaction/task/CompactionTaskStatusStore.java', 'java/compaction-core/src/main/java/sleeper/compaction/job/CompactionJobStatusStore.java', 'java/ingest/ingest-status-store/src/main/java/sleeper/ingest/job/status/DynamoDBIngestJobStatusStore.java', 'java/compaction-status-store/src/main/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStore.java', 'java/compaction-status-store/src/main/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStoreCreator.java', 'java/compaction-status-store/src/test/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStoreCreatorIT.java', 'java/compaction-job-execution/src/test/java/sleeper/compaction/jobexecution/testutils/CompactSortedFilesTestUtils.java', 'java/compaction-status-store/src/test/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStoreCreatorIT.java', 'java/configuration/src/main/java/sleeper/configuration/properties/SleeperProperties.java', 'java/compaction-status-store/src/main/java/sleeper/compaction/status/task/DynamoDBCompactionTaskStatusStore.java', 'java/compaction-status-store/src/main/java/sleeper/compaction/status/job/DynamoDBCompactionJobStatusStoreCreator.java']