Issue 8365 - Fix compaction commit batcher error handling

java/compaction/compaction-core/src/main/java/sleeper/compaction/core/job/commit/CompactionCommitMessageSerDe.java

@@ -17,11 +17,18 @@
 
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import sleeper.core.statestore.FileReferenceSerDe;
 import sleeper.core.util.GsonConfig;
 
+import java.util.Optional;
+
 public class CompactionCommitMessageSerDe {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(CompactionCommitMessageSerDe.class);
+
     private final Gson gson;
     private final Gson gsonPrettyPrint;
 
@@ -40,9 +47,15 @@ public String toJsonPrettyPrint(CompactionCommitMessage message) {
         return gsonPrettyPrint.toJson(message);
     }
 
-    public CompactionCommitMessageHandle fromJsonWithCallbackOnFail(String json, Runnable callbackOnFail) {
-        CompactionCommitMessage message = gson.fromJson(json, CompactionCommitMessage.class);
-        return new CompactionCommitMessageHandle(message.tableId(), message.request(), callbackOnFail);
+    public Optional<CompactionCommitMessageHandle> fromJsonWithCallbackOnFail(String json, Runnable callbackOnFail) {
+        try {
+            CompactionCommitMessage message = gson.fromJson(json, CompactionCommitMessage.class);
+            return Optional.of(new CompactionCommitMessageHandle(message.tableId(), message.request(), callbackOnFail));
+        } catch (RuntimeException e) {
+            LOGGER.error("Found invalid compaction commit message: {}", json, e);
+            callbackOnFail.run();
+            return Optional.empty();
+        }
     }
 
 }

java/compaction/compaction-core/src/test/java/sleeper/compaction/core/job/commit/CompactionCommitMessageSerDeTest.java

@@ -30,7 +30,9 @@
 import sleeper.core.table.TableFilePaths;
 import sleeper.core.table.TableIdGenerator;
 
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.IntStream;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -68,7 +70,7 @@ void shouldSerialiseCompactionCommitRequest() {
 
         // When
         String json = serDe.toJson(message);
-        CompactionCommitMessageHandle found = serDe.fromJsonWithCallbackOnFail(json, callbackOnFail);
+        CompactionCommitMessageHandle found = serDe.fromJsonWithCallbackOnFail(json, callbackOnFail).orElseThrow();
 
         // Then
         assertThat(found).isEqualTo(new CompactionCommitMessageHandle("test-table", filesRequest, callbackOnFail));
@@ -84,7 +86,7 @@ void shouldSerialiseRepresentativeCompactionCommitRequest() {
 
         // When
         String json = serDe.toJson(message);
-        CompactionCommitMessageHandle found = serDe.fromJsonWithCallbackOnFail(json, callbackOnFail);
+        CompactionCommitMessageHandle found = serDe.fromJsonWithCallbackOnFail(json, callbackOnFail).orElseThrow();
 
         // Then
         assertThat(found).isEqualTo(new CompactionCommitMessageHandle(tableId, filesRequest, callbackOnFail));
@@ -100,6 +102,21 @@ void shouldGenerateExpectedJson() {
                 .withName("example-compaction-commit", ".json"));
     }
 
+    @Test
+    void shouldTrackDeserialisationFailure() {
+        // Given
+        List<Boolean> callbackResults = new ArrayList<>();
+        Runnable callbackOnFail = () -> callbackResults.add(true);
+        String json = "not-a-commit";
+
+        // When
+        Optional<CompactionCommitMessageHandle> found = serDe.fromJsonWithCallbackOnFail(json, callbackOnFail);
+
+        // Then
+        assertThat(found).isEmpty();
+        assertThat(callbackResults).containsExactly(true);
+    }
+
     private ReplaceFileReferencesRequest createExampleFilesRequestOfRepresentativeSize() {
         PartitionTree partitions = new PartitionsBuilder(schema).singlePartition(exampleUUID("partn", 0)).buildTree();
         List<String> inputFiles = IntStream.rangeClosed(1, 11)

java/deployment/cdk/src/main/java/sleeper/cdk/stack/compaction/CompactionJobResources.java

@@ -335,6 +335,7 @@ private void lambdaToBatchUpCompactionCommits(
                 .batchSize(instanceProperties.getInt(COMPACTION_COMMIT_BATCH_SIZE))
                 .maxBatchingWindow(Duration.seconds(instanceProperties.getInt(COMPACTION_COMMIT_BATCHING_WINDOW_IN_SECONDS)))
                 .maxConcurrency(instanceProperties.getIntOrNull(COMPACTION_COMMIT_BATCHER_LAMBDA_CONCURRENCY_MAXIMUM))
+                .reportBatchItemFailures(true)
                 .build());
         coreStacks.grantSendStateStoreCommits(function);
     }

java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json

@@ -17332,6 +17332,9 @@
                 "Arn"
               ]
             },
+            "FunctionResponseTypes": [
+              "ReportBatchItemFailures"
+            ],
             "Tags": [
               {
                 "Key": "DeploymentStack",

java/statestore-lambda/src/main/java/sleeper/statestore/lambda/committer/CompactionCommitBatcherLambda.java

@@ -20,7 +20,6 @@
 import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse;
 import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse.BatchItemFailure;
 import com.amazonaws.services.lambda.runtime.events.SQSEvent;
-import com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.s3.S3Client;
@@ -64,17 +63,14 @@ public CompactionCommitBatcherLambda(CompactionCommitBatcher batcher) {
     public SQSBatchResponse handleRequest(SQSEvent event, Context context) {
         List<BatchItemFailure> failures = new ArrayList<>();
         List<CompactionCommitMessageHandle> requests = event.getRecords().stream()
-                .map(message -> readMessageTrackingFailure(message, failures))
+                .flatMap(message -> serDe.fromJsonWithCallbackOnFail(message.getBody(),
+                        () -> failures.add(new BatchItemFailure(message.getMessageId())))
+                        .stream())
                 .toList();
         batcher.sendBatch(requests);
         return new SQSBatchResponse(failures);
     }
 
-    private CompactionCommitMessageHandle readMessageTrackingFailure(SQSMessage message, List<BatchItemFailure> failures) {
-        return serDe.fromJsonWithCallbackOnFail(message.getBody(),
-                () -> failures.add(new BatchItemFailure(message.getMessageId())));
-    }
-
     /**
      * Creates the batcher used to send requests to the state store committer queue via SQS.
      *

java/statestore-lambda/src/test/java/sleeper/statestore/lambda/committer/CompactionCommitBatcherLambdaIT.java

@@ -85,7 +85,30 @@ void shouldFailWhenUnableToGetToTheSQSQueue() {
         assertThat(response.getBatchItemFailures())
                 .extracting(BatchItemFailure::getItemIdentifier)
                 .containsExactly(sqsMessage.getMessageId());
+    }
+
+    @Test
+    void shouldFailWhenInvalidAndValidMessagesAreMixedInOneBatch() {
+        // Given
+        ReplaceFileReferencesRequest filesRequest = createFilesRequest();
+        List<SQSMessage> messages = List.of(
+                createMessageWithId("1-valid", new CompactionCommitMessage("table-1", filesRequest)),
+                createMessageWithId("2-invalid", "not-a-commit"),
+                createMessageWithId("3-valid", new CompactionCommitMessage("table-2", filesRequest)),
+                createMessageWithId("4-invalid", "not-a-commit"));
 
+        // When
+        SQSBatchResponse response = lambda().handleRequest(createEvent(messages), null);
+
+        // Then
+        assertThat(receiveCommitRequests()).containsExactly(
+                StateStoreCommitRequest.create("table-1",
+                        new ReplaceFileReferencesTransaction(List.of(filesRequest))),
+                StateStoreCommitRequest.create("table-2",
+                        new ReplaceFileReferencesTransaction(List.of(filesRequest))));
+        assertThat(response.getBatchItemFailures())
+                .extracting(BatchItemFailure::getItemIdentifier)
+                .containsExactly("2-invalid", "4-invalid");
     }
 
     private ReplaceFileReferencesRequest createFilesRequest() {
@@ -109,15 +132,27 @@ private SQSEvent createEvent(String tableId, ReplaceFileReferencesRequest reques
     }
 
     private SQSEvent createEvent(SQSMessage... messages) {
+        return createEvent(List.of(messages));
+    }
+
+    private SQSEvent createEvent(List<SQSMessage> messages) {
         SQSEvent event = new SQSEvent();
-        event.setRecords(List.of(messages));
+        event.setRecords(messages);
         return event;
     }
 
     private SQSMessage createMessage(String tableId, ReplaceFileReferencesRequest request) {
+        return createMessageWithId(UUID.randomUUID().toString(), new CompactionCommitMessage(tableId, request));
+    }
+
+    private SQSMessage createMessageWithId(String id, CompactionCommitMessage commit) {
+        return createMessageWithId(id, compactionSerDe.toJson(commit));
+    }
+
+    private SQSMessage createMessageWithId(String id, String body) {
         SQSMessage message = new SQSMessage();
-        message.setMessageId(UUID.randomUUID().toString());
-        message.setBody(compactionSerDe.toJson(new CompactionCommitMessage(tableId, request)));
+        message.setMessageId(id);
+        message.setBody(body);
         return message;
     }
 

java/system-test/system-test-drivers/src/test/java/sleeper/systemtest/drivers/compaction/CompactionFakeCommitIT.java

@@ -96,7 +96,7 @@ private Set<CompactionCommitMessageHandle> drainCommitQueue() {
     }
 
     private CompactionCommitMessageHandle readCommitMessage(Message message) {
-        return serDe.fromJsonWithCallbackOnFail(message.body(), null);
+        return serDe.fromJsonWithCallbackOnFail(message.body(), null).orElseThrow();
     }
 
     private CompactionCommitMessageHandle handle(ReplaceFileReferencesRequest request) {

java/system-test/system-test-suite/src/test/java/sleeper/systemtest/suite/LambdaBatchFailuresST.java

@@ -102,7 +102,6 @@ void shouldReportFailureCreatingCompactionJobsForSomeTables(SleeperDsl sleeper)
     }
 
     @Test
-    @Disabled("Failing until issue https://github.com/gchq/sleeper/issues/8365")
     void shouldReportFailureCommittingSomeCompactionJobs(SleeperDsl sleeper) {
         // Given a Sleeper table with 5 compaction jobs ready to be committed to the state store
         sleeper.tables().createWithProperties("test", DEFAULT_SCHEMA, Map.of(