Repository: gchq/sleeper
Issue: Ingest batcher state store tests fail intermittently

A test failed to query pending records and assert the sort order. No records were returned at all, when 3 records were added to the store.

It's not clear why this would happen. We can update this issue when it happens and try to figure it out.

Here are several times this happened in DynamoDBIngestBatcherStoreIT$OrderFilesReturnedFromStore.shouldListPendingFilesInOrderRequestsReceivedOldestFirst:

https://github.com/gchq/sleeper/actions/runs/5344363020/jobs/9688654804
https://github.com/gchq/sleeper/runs/14467112289

https://github.com/gchq/sleeper/actions/runs/5389911405/jobs/9784528532
https://github.com/gchq/sleeper/runs/14589403331

Here it happened in DynamoDBIngestBatcherStoreIT$AssignFilesToJobs.shouldFailToAssignFileWhenAssignmentAlreadyExists:

https://github.com/gchq/sleeper/actions/runs/5654678408/job/15318234077
https://github.com/gchq/sleeper/runs/15318336515

Here it happened in DynamoDBIngestBatcherStoreIT$AssignFilesToJobs.shouldFailToAssignFilesWhenOneIsAlreadyAssigned, with just one of the files missing:

https://github.com/mh674563/sleeper/actions/runs/5586589682/jobs/10210939461
https://github.com/mh674563/sleeper/actions/runs/5586589706/attempts/1

Here it happened in DynamoDBIngestBatcherStoreIT$OrderFilesReturnedFromStore.shouldReportAllFilesInOrderRequestsReceivedMostRecentFirst:

https://github.com/gchq/sleeper/actions/runs/5760737290/attempts/1

Here are some transaction cancellation failures:
https://github.com/gchq/sleeper/actions/runs/5521500088/jobs/10069693700
https://github.com/gchq/sleeper/runs/14949826850
https://github.com/gchq/sleeper/actions/runs/5654361346/job/15317281795?pr=1043


Changed files:
java/ingest/ingest-batcher-store/src/test/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStoreTestBase.java
java/ingest/ingest-batcher-store/src/test/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStoreIT.java
java/ingest/ingest-batcher-store/src/main/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStore.java

diff --git a/java/ingest/ingest-batcher-store/src/main/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStore.java b/java/ingest/ingest-batcher-store/src/main/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStore.java
index 0441358ab..ac1c48f3e 100644
--- a/java/ingest/ingest-batcher-store/src/main/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStore.java
+++ b/java/ingest/ingest-batcher-store/src/main/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStore.java
@@ -18,15 +18,18 @@ package sleeper.ingest.batcher.store;
 
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
 import com.amazonaws.services.dynamodbv2.model.BatchWriteItemRequest;
+import com.amazonaws.services.dynamodbv2.model.ConsumedCapacity;
 import com.amazonaws.services.dynamodbv2.model.Delete;
 import com.amazonaws.services.dynamodbv2.model.DeleteRequest;
 import com.amazonaws.services.dynamodbv2.model.Put;
 import com.amazonaws.services.dynamodbv2.model.PutItemRequest;
+import com.amazonaws.services.dynamodbv2.model.PutItemResult;
 import com.amazonaws.services.dynamodbv2.model.QueryRequest;
 import com.amazonaws.services.dynamodbv2.model.ReturnConsumedCapacity;
 import com.amazonaws.services.dynamodbv2.model.ScanRequest;
 import com.amazonaws.services.dynamodbv2.model.TransactWriteItem;
 import com.amazonaws.services.dynamodbv2.model.TransactWriteItemsRequest;
+import com.amazonaws.services.dynamodbv2.model.TransactWriteItemsResult;
 import com.amazonaws.services.dynamodbv2.model.TransactionCanceledException;
 import com.amazonaws.services.dynamodbv2.model.WriteRequest;
 import org.slf4j.Logger;
@@ -41,6 +44,7 @@ import sleeper.ingest.batcher.IngestBatcherStore;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -73,10 +77,12 @@ public class DynamoDBIngestBatcherStore implements IngestBatcherStore {
 
     @Override
     public void addFile(FileIngestRequest fileIngestRequest) {
-        dynamoDB.putItem(new PutItemRequest()
+        PutItemResult result = dynamoDB.putItem(new PutItemRequest()
                 .withTableName(requestsTableName)
                 .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
                 .withItem(DynamoDBIngestRequestFormat.createRecord(tablePropertiesProvider, fileIngestRequest)));
+        LOGGER.debug("Put request to ingest file {} to table {}, capacity consumed = {}",
+                fileIngestRequest.getFile(), fileIngestRequest.getTableName(), result.getConsumedCapacity().getCapacityUnits());
     }
 
     @Override
@@ -85,7 +91,8 @@ public class DynamoDBIngestBatcherStore implements IngestBatcherStore {
         for (int i = 0; i < filesInJob.size(); i += 50) {
             List<FileIngestRequest> filesInBatch = filesInJob.subList(i, Math.min(i + 50, filesInJob.size()));
             try {
-                dynamoDB.transactWriteItems(new TransactWriteItemsRequest()
+                TransactWriteItemsRequest request = new TransactWriteItemsRequest()
+                        .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
                         .withTransactItems(filesInBatch.stream()
                                 .flatMap(file -> Stream.of(
                                         new TransactWriteItem().withDelete(new Delete()
@@ -99,11 +106,18 @@ public class DynamoDBIngestBatcherStore implements IngestBatcherStore {
                                                         tablePropertiesProvider, file.toBuilder().jobId(jobId).build()))
                                                 .withConditionExpression("attribute_not_exists(#filepath)")
                                                 .withExpressionAttributeNames(Map.of("#filepath", FILE_PATH))))
-                                ).collect(Collectors.toList())));
+                                ).collect(Collectors.toList()));
+                TransactWriteItemsResult result = dynamoDB.transactWriteItems(request);
+                List<ConsumedCapacity> consumedCapacity = Optional.ofNullable(result.getConsumedCapacity()).orElse(List.of());
+                double totalConsumed = consumedCapacity.stream().mapToDouble(ConsumedCapacity::getCapacityUnits).sum();
+                LOGGER.debug("Assigned {} files to job {}, capacity consumed = {}",
+                        filesInBatch.size(), jobId, totalConsumed);
                 assignedFiles.addAll(filesInBatch);
             } catch (TransactionCanceledException e) {
                 LOGGER.error("{} files could not be batched, leaving them for next batcher run.", filesInBatch.size());
-                LOGGER.error("Cancellation reasons: {}", e.getCancellationReasons(), e);
+                long numFailures = e.getCancellationReasons().stream()
+                        .filter(reason -> !"None".equals(reason.getCode())).count();
+                LOGGER.error("Cancellation reasons ({} failures): {}", numFailures, e.getCancellationReasons(), e);
             } catch (RuntimeException e) {
                 LOGGER.error("{} files could not be batched, leaving them for next batcher run.", filesInBatch.size(), e);
             }
diff --git a/java/ingest/ingest-batcher-store/src/test/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStoreIT.java b/java/ingest/ingest-batcher-store/src/test/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStoreIT.java
index f721427dc..832ef8abf 100644
--- a/java/ingest/ingest-batcher-store/src/test/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStoreIT.java
+++ b/java/ingest/ingest-batcher-store/src/test/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStoreIT.java
@@ -256,9 +256,9 @@ public class DynamoDBIngestBatcherStoreIT extends DynamoDBIngestBatcherStoreTest
             // Given
             // Transaction limit is 100. 2 transactions are performed per job, so send 50 files
             List<FileIngestRequest> fileIngestRequests = IntStream.range(0, 50)
-                    .mapToObj(i -> fileRequest().file("test-bucket/file-" + i + ".parquet").build())
+                    .mapToObj(i -> fileRequest().build())
                     .collect(Collectors.toUnmodifiableList());
-            fileIngestRequests.forEach(store::addFile);
+            addFiles(fileIngestRequests); // Add files transactionally as DynamoDB local is inconsistent otherwise
 
             // When
             store.assignJobGetAssigned("test-job", fileIngestRequests);
@@ -272,9 +272,9 @@ public class DynamoDBIngestBatcherStoreIT extends DynamoDBIngestBatcherStoreTest
             // Given
             // Transaction limit is 100. 2 transactions are performed per job, so send 51 files
             List<FileIngestRequest> fileIngestRequests = IntStream.range(0, 51)
-                    .mapToObj(i -> fileRequest().file("test-bucket/file-" + i + ".parquet").build())
+                    .mapToObj(i -> fileRequest().build())
                     .collect(Collectors.toUnmodifiableList());
-            fileIngestRequests.forEach(store::addFile);
+            addFiles(fileIngestRequests); // Add files transactionally as DynamoDB local is inconsistent otherwise
 
             // When
             store.assignJobGetAssigned("test-job", fileIngestRequests);
diff --git a/java/ingest/ingest-batcher-store/src/test/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStoreTestBase.java b/java/ingest/ingest-batcher-store/src/test/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStoreTestBase.java
index 9386f9987..e01007d18 100644
--- a/java/ingest/ingest-batcher-store/src/test/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStoreTestBase.java
+++ b/java/ingest/ingest-batcher-store/src/test/java/sleeper/ingest/batcher/store/DynamoDBIngestBatcherStoreTestBase.java
@@ -15,6 +15,10 @@
  */
 package sleeper.ingest.batcher.store;
 
+import com.amazonaws.services.dynamodbv2.model.Put;
+import com.amazonaws.services.dynamodbv2.model.ReturnConsumedCapacity;
+import com.amazonaws.services.dynamodbv2.model.TransactWriteItem;
+import com.amazonaws.services.dynamodbv2.model.TransactWriteItemsRequest;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 
@@ -23,9 +27,11 @@ import sleeper.configuration.properties.table.FixedTablePropertiesProvider;
 import sleeper.configuration.properties.table.TableProperties;
 import sleeper.configuration.properties.table.TablePropertiesProvider;
 import sleeper.dynamodb.tools.DynamoDBTestBase;
+import sleeper.ingest.batcher.FileIngestRequest;
 import sleeper.ingest.batcher.IngestBatcherStore;
 
 import java.util.List;
+import java.util.stream.Collectors;
 
 import static sleeper.configuration.properties.InstancePropertiesTestHelper.createTestInstanceProperties;
 import static sleeper.configuration.properties.instance.CommonProperty.ID;
@@ -55,4 +61,14 @@ public class DynamoDBIngestBatcherStoreTestBase extends DynamoDBTestBase {
     void tearDown() {
         DynamoDBIngestBatcherStoreCreator.tearDown(instanceProperties, dynamoDBClient);
     }
+
+    protected void addFiles(List<FileIngestRequest> fileIngestRequests) {
+        dynamoDBClient.transactWriteItems(new TransactWriteItemsRequest()
+                .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
+                .withTransactItems(fileIngestRequests.stream()
+                        .map(file -> new TransactWriteItem().withPut(new Put()
+                                .withTableName(requestsTableName)
+                                .withItem(DynamoDBIngestRequestFormat.createRecord(tablePropertiesProvider, file)))
+                        ).collect(Collectors.toList())));
+    }
 }