DynamoDBIngestJobStatusStore is used directly even when store may not be enabled
It's not possible to disable the ingest job status store fully because DynamoDBIngestJobStatusStore is used directly in several places. IngestJobStatusStoreFactory should be used instead.
diff --git a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/BulkImportStarterLambda.java b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/BulkImportStarterLambda.java
index 321bb58ec..fade1c2a4 100644
--- a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/BulkImportStarterLambda.java
+++ b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/BulkImportStarterLambda.java
@@ -33,7 +33,7 @@ import sleeper.bulkimport.starter.executor.PlatformExecutor;
 import sleeper.configuration.properties.InstanceProperties;
 import sleeper.configuration.properties.table.TablePropertiesProvider;
 import sleeper.ingest.job.status.IngestJobStatusStore;
-import sleeper.ingest.status.store.job.DynamoDBIngestJobStatusStore;
+import sleeper.ingest.status.store.job.IngestJobStatusStoreFactory;
 import sleeper.statestore.StateStoreProvider;
 import sleeper.utils.HadoopPathUtils;
 
@@ -69,7 +69,7 @@ public class BulkImportStarterLambda implements RequestHandler<SQSEvent, Void> {
         PlatformExecutor platformExecutor = PlatformExecutor.fromEnvironment(
                 instanceProperties, tablePropertiesProvider);
         hadoopConfig = new Configuration();
-        ingestJobStatusStore = new DynamoDBIngestJobStatusStore(dynamo, instanceProperties);
+        ingestJobStatusStore = IngestJobStatusStoreFactory.getStatusStore(dynamo, instanceProperties);
         executor = new BulkImportExecutor(instanceProperties, tablePropertiesProvider,
                 new StateStoreProvider(dynamo, instanceProperties),
                 ingestJobStatusStore, s3, platformExecutor, Instant::now);
diff --git a/java/bulk-import/bulk-import-starter/src/test/java/sleeper/bulkimport/starter/executor/BulkImportExecutorIT.java b/java/bulk-import/bulk-import-starter/src/test/java/sleeper/bulkimport/starter/executor/BulkImportExecutorIT.java
index f9832dfbd..493a8a65c 100644
--- a/java/bulk-import/bulk-import-starter/src/test/java/sleeper/bulkimport/starter/executor/BulkImportExecutorIT.java
+++ b/java/bulk-import/bulk-import-starter/src/test/java/sleeper/bulkimport/starter/executor/BulkImportExecutorIT.java
@@ -43,6 +43,7 @@ import sleeper.core.schema.type.StringType;
 import sleeper.ingest.job.status.IngestJobStatusStore;
 import sleeper.ingest.status.store.job.DynamoDBIngestJobStatusStore;
 import sleeper.ingest.status.store.job.DynamoDBIngestJobStatusStoreCreator;
+import sleeper.ingest.status.store.job.IngestJobStatusStoreFactory;
 import sleeper.statestore.FixedStateStoreProvider;
 import sleeper.statestore.StateStoreProvider;
 
@@ -93,7 +94,7 @@ class BulkImportExecutorIT {
     private final TableProperties tableProperties = createTestTableProperties(instanceProperties, SCHEMA);
     private final String bucketName = UUID.randomUUID().toString();
     private final String tableName = "myTable";
-    private final IngestJobStatusStore ingestJobStatusStore = new DynamoDBIngestJobStatusStore(dynamoDB, instanceProperties);
+    private final IngestJobStatusStore ingestJobStatusStore = IngestJobStatusStoreFactory.getStatusStore(dynamoDB, instanceProperties);
 
     @BeforeEach
     void setup() {
diff --git a/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/status/store/job/DynamoDBIngestJobStatusStore.java b/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/status/store/job/DynamoDBIngestJobStatusStore.java
index df78b056b..1c3cb3118 100644
--- a/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/status/store/job/DynamoDBIngestJobStatusStore.java
+++ b/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/status/store/job/DynamoDBIngestJobStatusStore.java
@@ -59,11 +59,7 @@ public class DynamoDBIngestJobStatusStore implements IngestJobStatusStore {
     private final String statusTableName;
     private final DynamoDBIngestJobStatusFormat format;
 
-    public DynamoDBIngestJobStatusStore(AmazonDynamoDB dynamoDB, InstanceProperties properties) {
-        this(dynamoDB, properties, Instant::now);
-    }
-
-    public DynamoDBIngestJobStatusStore(AmazonDynamoDB dynamoDB, InstanceProperties properties, Supplier<Instant> getTimeNow) {
+    DynamoDBIngestJobStatusStore(AmazonDynamoDB dynamoDB, InstanceProperties properties, Supplier<Instant> getTimeNow) {
         this.dynamoDB = dynamoDB;
         this.statusTableName = jobStatusTableName(properties.get(ID));
         int timeToLiveInSeconds = properties.getInt(UserDefinedInstanceProperty.INGEST_JOB_STATUS_TTL_IN_SECONDS);
diff --git a/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/status/store/job/IngestJobStatusStoreFactory.java b/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/status/store/job/IngestJobStatusStoreFactory.java
index ca2a2ea95..43754c21d 100644
--- a/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/status/store/job/IngestJobStatusStoreFactory.java
+++ b/java/ingest/ingest-status-store/src/main/java/sleeper/ingest/status/store/job/IngestJobStatusStoreFactory.java
@@ -21,6 +21,9 @@ import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
 import sleeper.configuration.properties.InstanceProperties;
 import sleeper.ingest.job.status.IngestJobStatusStore;
 
+import java.time.Instant;
+import java.util.function.Supplier;
+
 import static sleeper.configuration.properties.UserDefinedInstanceProperty.INGEST_STATUS_STORE_ENABLED;
 
 public class IngestJobStatusStoreFactory {
@@ -29,8 +32,12 @@ public class IngestJobStatusStoreFactory {
     }
 
     public static IngestJobStatusStore getStatusStore(AmazonDynamoDB dynamoDB, InstanceProperties properties) {
+        return getStatusStore(dynamoDB, properties, Instant::now);
+    }
+
+    public static IngestJobStatusStore getStatusStore(AmazonDynamoDB dynamoDB, InstanceProperties properties, Supplier<Instant> getTimeNow) {
         if (properties.getBoolean(INGEST_STATUS_STORE_ENABLED)) {
-            return new DynamoDBIngestJobStatusStore(dynamoDB, properties);
+            return new DynamoDBIngestJobStatusStore(dynamoDB, properties, getTimeNow);
         } else {
             return IngestJobStatusStore.NONE;
         }
diff --git a/java/ingest/ingest-status-store/src/test/java/sleeper/ingest/status/store/testutils/DynamoDBIngestJobStatusStoreTestBase.java b/java/ingest/ingest-status-store/src/test/java/sleeper/ingest/status/store/testutils/DynamoDBIngestJobStatusStoreTestBase.java
index 5887a8bc5..b67d11bd2 100644
--- a/java/ingest/ingest-status-store/src/test/java/sleeper/ingest/status/store/testutils/DynamoDBIngestJobStatusStoreTestBase.java
+++ b/java/ingest/ingest-status-store/src/test/java/sleeper/ingest/status/store/testutils/DynamoDBIngestJobStatusStoreTestBase.java
@@ -85,7 +85,7 @@ public class DynamoDBIngestJobStatusStoreTestBase extends DynamoDBTestBase {
     }
 
     protected IngestJobStatusStore storeWithUpdateTimes(Instant... updateTimes) {
-        return new DynamoDBIngestJobStatusStore(dynamoDBClient, instanceProperties,
+        return IngestJobStatusStoreFactory.getStatusStore(dynamoDBClient, instanceProperties,
                 Arrays.stream(updateTimes).iterator()::next);
     }
 
diff --git a/java/system-test/src/main/java/sleeper/systemtest/ingest/batcher/InvokeIngestBatcher.java b/java/system-test/src/main/java/sleeper/systemtest/ingest/batcher/InvokeIngestBatcher.java
index 8bf2fbb13..1dbca2e17 100644
--- a/java/system-test/src/main/java/sleeper/systemtest/ingest/batcher/InvokeIngestBatcher.java
+++ b/java/system-test/src/main/java/sleeper/systemtest/ingest/batcher/InvokeIngestBatcher.java
@@ -32,7 +32,7 @@ import sleeper.ingest.batcher.IngestBatcherStore;
 import sleeper.ingest.batcher.store.DynamoDBIngestBatcherStore;
 import sleeper.ingest.job.status.IngestJobStatus;
 import sleeper.ingest.job.status.IngestJobStatusStore;
-import sleeper.ingest.status.store.job.DynamoDBIngestJobStatusStore;
+import sleeper.ingest.status.store.job.IngestJobStatusStoreFactory;
 import sleeper.job.common.QueueMessageCount;
 import sleeper.systemtest.util.InvokeSystemTestLambda;
 import sleeper.systemtest.util.WaitForQueueEstimate;
@@ -79,7 +79,7 @@ public class InvokeIngestBatcher {
                 tablePropertiesUpdater(tableProperties, s3),
                 new SendFilesToIngestBatcher(instanceProperties, tableProperties, sourceBucketName,
                         batcherStore, sqs, lambda),
-                batcherStore, new DynamoDBIngestJobStatusStore(dynamoDB, instanceProperties),
+                batcherStore, IngestJobStatusStoreFactory.getStatusStore(dynamoDB, instanceProperties),
                 InvokeSystemTestLambda.client(lambda, instanceProperties),
                 QueueMessageCount.withSqsClient(sqs));
     }
['java/system-test/src/main/java/sleeper/systemtest/ingest/batcher/InvokeIngestBatcher.java', 'java/ingest/ingest-status-store/src/main/java/sleeper/ingest/status/store/job/IngestJobStatusStoreFactory.java', 'java/bulk-import/bulk-import-starter/src/test/java/sleeper/bulkimport/starter/executor/BulkImportExecutorIT.java', 'java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/BulkImportStarterLambda.java', 'java/ingest/ingest-status-store/src/main/java/sleeper/ingest/status/store/job/DynamoDBIngestJobStatusStore.java', 'java/ingest/ingest-status-store/src/test/java/sleeper/ingest/status/store/testutils/DynamoDBIngestJobStatusStoreTestBase.java']