Issue 8363 - Report batch item failures

java/compaction/compaction-job-creation-lambda/pom.xml

@@ -64,6 +64,14 @@
             <artifactId>common-invoke-tables</artifactId>
             <version>${project.parent.version}</version>
         </dependency>
+        <!-- Test Dependencies -->
+        <dependency>
+            <groupId>sleeper</groupId>
+            <artifactId>core</artifactId>
+            <version>${project.parent.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

java/compaction/compaction-job-creation-lambda/src/main/java/sleeper/compaction/job/creation/lambda/CreateCompactionJobsLambda.java

@@ -37,6 +37,7 @@
 import sleeper.core.properties.table.TableProperties;
 import sleeper.core.properties.table.TablePropertiesProvider;
 import sleeper.core.statestore.StateStoreProvider;
+import sleeper.core.table.TableNotFoundException;
 import sleeper.core.util.LoggedDuration;
 import sleeper.core.util.ObjectFactory;
 import sleeper.core.util.ObjectFactoryException;
@@ -87,6 +88,12 @@ public CreateCompactionJobsLambda() throws ObjectFactoryException {
                 objectFactory, instanceProperties, tablePropertiesProvider, stateStoreProvider, s3Client, sqsClient);
     }
 
+    public CreateCompactionJobsLambda(TablePropertiesProvider tablePropertiesProvider, PropertiesReloader propertiesReloader, CreateCompactionJobs createJobs) {
+        this.tablePropertiesProvider = tablePropertiesProvider;
+        this.propertiesReloader = propertiesReloader;
+        this.createJobs = createJobs;
+    }
+
     public SQSBatchResponse handleRequest(SQSEvent event, Context context) {
         Instant startTime = Instant.now();
         LOGGER.info("Lambda started at {}", startTime);
@@ -102,6 +109,8 @@ public SQSBatchResponse handleRequest(SQSEvent event, Context context) {
                 TableProperties tableProperties = tablePropertiesProvider.getById(tableId);
                 LOGGER.info("Received {} messages for table {}", tableMessages.size(), tableProperties.getStatus());
                 createJobs.createJobsWithStrategy(tableProperties);
+            } catch (TableNotFoundException e) {
+                LOGGER.warn("Ignoring non-existent table {}", tableId);
             } catch (RuntimeException | ObjectFactoryException e) {
                 LOGGER.error("Failed creating jobs for table {}", tableId, e);
                 tableMessages.stream()

java/compaction/compaction-job-creation-lambda/src/test/java/sleeper/compaction/job/creation/lambda/CreateCompactionJobsLambdaTest.java

@@ -0,0 +1,176 @@
+/*
+ * Copyright 2022-2026 Crown Copyright
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package sleeper.compaction.job.creation.lambda;
+
+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.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import sleeper.compaction.core.job.CompactionJob;
+import sleeper.compaction.core.job.CompactionJobFactory;
+import sleeper.compaction.core.job.creation.CreateCompactionJobBatches;
+import sleeper.compaction.core.job.creation.CreateCompactionJobBatches.BatchJobsWriter;
+import sleeper.compaction.core.job.creation.CreateCompactionJobBatches.BatchMessageSender;
+import sleeper.compaction.core.job.creation.CreateCompactionJobs;
+import sleeper.compaction.core.job.creation.strategy.impl.BasicCompactionStrategy;
+import sleeper.core.properties.PropertiesReloader;
+import sleeper.core.properties.instance.InstanceProperties;
+import sleeper.core.properties.table.TableProperties;
+import sleeper.core.properties.testutils.FixedTablePropertiesProvider;
+import sleeper.core.statestore.FileReference;
+import sleeper.core.statestore.FileReferenceFactory;
+import sleeper.core.statestore.StateStore;
+import sleeper.core.statestore.StateStoreProvider;
+import sleeper.core.statestore.commit.StateStoreCommitRequestSender;
+import sleeper.core.statestore.testutils.InMemoryTransactionLogStateStore;
+import sleeper.core.statestore.testutils.InMemoryTransactionLogsPerTable;
+import sleeper.core.util.ObjectFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static sleeper.core.properties.table.TableProperty.COMPACTION_FILES_BATCH_SIZE;
+import static sleeper.core.properties.table.TableProperty.COMPACTION_STRATEGY_CLASS;
+import static sleeper.core.properties.table.TableProperty.TABLE_ID;
+import static sleeper.core.properties.testutils.InstancePropertiesTestHelper.createTestInstanceProperties;
+import static sleeper.core.properties.testutils.TablePropertiesTestHelper.createTestTableProperties;
+import static sleeper.core.schema.SchemaTestHelper.createSchemaWithKey;
+import static sleeper.core.statestore.testutils.StateStoreUpdatesWrapper.update;
+import static sleeper.core.testutils.SupplierTestHelper.supplyNumberedIdsWithPrefix;
+import static sleeper.core.testutils.SupplierTestHelper.timePassesAMinuteAtATime;
+
+public class CreateCompactionJobsLambdaTest {
+
+    private final InstanceProperties instanceProperties = createTestInstanceProperties();
+    private final TableProperties tableProperties = createTestTableProperties(instanceProperties, createSchemaWithKey("key"));
+    private final InMemoryTransactionLogsPerTable transactionLogs = new InMemoryTransactionLogsPerTable();
+    private final StateStoreProvider stateStoreProvider = InMemoryTransactionLogStateStore.createProvider(instanceProperties, transactionLogs);
+    private final List<CompactionJob> jobsCreated = new ArrayList<>();
+
+    @BeforeEach
+    void setUp() {
+        tableProperties.set(COMPACTION_STRATEGY_CLASS, BasicCompactionStrategy.class.getName());
+        tableProperties.setNumber(COMPACTION_FILES_BATCH_SIZE, 2);
+    }
+
+    @Test
+    void shouldCreateJob() {
+        // Given
+        update(stateStore()).initialise(tableProperties);
+        FileReference file1 = fileFactory().rootFile("file1", 10L);
+        FileReference file2 = fileFactory().rootFile("file2", 20L);
+        update(stateStore()).addFiles(List.of(file1, file2));
+
+        // When
+        SQSBatchResponse response = invokeForSingleTable();
+
+        // Then
+        assertThat(jobsCreated).containsExactly(
+                compactionFactory().createCompactionJob("job1", List.of(file1, file2), "root"));
+        assertThat(response.getBatchItemFailures()).isEmpty();
+    }
+
+    @Test
+    void shouldFailWhenTableIsUnreadable() {
+        // Given
+        transactionLogs.forTable(tableProperties).getFilesLogStore().atStartOfReadTransactions(() -> {
+            throw new RuntimeException("The file transaction log is unreadable");
+        });
+
+        // When
+        SQSBatchResponse response = invokeForSingleTable();
+
+        // Then
+        assertThat(response.getBatchItemFailures()).singleElement()
+                .extracting(BatchItemFailure::getItemIdentifier)
+                .isEqualTo("test-message");
+    }
+
+    @Test
+    void shouldIgnoreTableThatWasDeleted() {
+        // When
+        SQSBatchResponse response = invokeForSingleTable("missing-table");
+
+        // Then
+        assertThat(response.getBatchItemFailures()).isEmpty();
+    }
+
+    private SQSBatchResponse invokeForSingleTable() {
+        return invokeForSingleTable(tableProperties.get(TABLE_ID));
+    }
+
+    private SQSBatchResponse invokeForSingleTable(String tableId) {
+        return lambda().handleRequest(event(
+                messageForTableWithId(tableId, "test-message")), null);
+    }
+
+    private CreateCompactionJobsLambda lambda() {
+        return new CreateCompactionJobsLambda(new FixedTablePropertiesProvider(tableProperties), PropertiesReloader.neverReload(), creator());
+    }
+
+    private SQSEvent event(SQSMessage... messages) {
+        SQSEvent event = new SQSEvent();
+        event.setRecords(List.of(messages));
+        return event;
+    }
+
+    private SQSMessage messageForTableWithId(String tableId, String messageId) {
+        SQSMessage message = new SQSMessage();
+        message.setBody(tableId);
+        message.setMessageId(messageId);
+        return message;
+    }
+
+    private CreateCompactionJobs creator() {
+        return new CreateCompactionJobs(instanceProperties, stateStoreProvider, batcher(), ObjectFactory.noUserJars(), supplyNumberedIdsWithPrefix("job")::get, new Random(0));
+    }
+
+    private CreateCompactionJobBatches batcher() {
+        return new CreateCompactionJobBatches(instanceProperties, jobsWriter(), batchSender(), stateStoreCommitSender(), supplyNumberedIdsWithPrefix("batch")::get, timePassesAMinuteAtATime());
+    }
+
+    private StateStore stateStore() {
+        return stateStoreProvider.getStateStore(tableProperties);
+    }
+
+    private FileReferenceFactory fileFactory() {
+        return FileReferenceFactory.from(instanceProperties, tableProperties, stateStore());
+    }
+
+    private CompactionJobFactory compactionFactory() {
+        return new CompactionJobFactory(instanceProperties, tableProperties);
+    }
+
+    private BatchJobsWriter jobsWriter() {
+        return (bucket, key, jobs) -> jobsCreated.addAll(jobs);
+    }
+
+    private BatchMessageSender batchSender() {
+        return request -> {
+        };
+    }
+
+    private StateStoreCommitRequestSender stateStoreCommitSender() {
+        return request -> {
+        };
+    }
+
+}

java/core/src/test/java/sleeper/core/statestore/testutils/InMemoryTransactionLogStore.java

@@ -167,6 +167,10 @@ public TransactionLogEntry getLastEntry() {
         return transactionEntries.get(transactionEntries.size() - 1);
     }
 
+    public List<TransactionLogEntry> getTransactionEntries() {
+        return transactionEntries;
+    }
+
     private void doStartOfAddTransaction() {
         startOfAdd.run();
     }

java/deployment/cdk/src/main/java/sleeper/cdk/stack/GarbageCollectorStack.java

@@ -131,6 +131,7 @@ public GarbageCollectorStack(
         handlerFunction.addEventSource(SqsEventSource.Builder.create(queue)
                 .batchSize(instanceProperties.getInt(GARBAGE_COLLECTOR_TABLE_BATCH_SIZE))
                 .maxConcurrency(instanceProperties.getIntOrNull(GARBAGE_COLLECTOR_LAMBDA_CONCURRENCY_MAXIMUM))
+                .reportBatchItemFailures(true)
                 .build());
         coreStacks.grantInvokeScheduled(triggerFunction, queue);
 

java/deployment/cdk/src/main/java/sleeper/cdk/stack/TableMetricsStack.java

@@ -115,6 +115,7 @@ public TableMetricsStack(
         tableMetricsPublisher.addEventSource(SqsEventSource.Builder.create(queue)
                 .batchSize(instanceProperties.getInt(METRICS_TABLE_BATCH_SIZE))
                 .maxConcurrency(instanceProperties.getIntOrNull(METRICS_LAMBDA_CONCURRENCY_MAXIMUM))
+                .reportBatchItemFailures(true)
                 .build());
 
         coreStacks.grantReadTablesStatus(tableMetricsTrigger);

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

@@ -204,6 +204,7 @@ private IFunction lambdaToCreateCompactionJobBatches(
         handlerFunction.addEventSource(SqsEventSource.Builder.create(jobCreationQueue)
                 .batchSize(instanceProperties.getInt(COMPACTION_JOB_CREATION_BATCH_SIZE))
                 .maxConcurrency(instanceProperties.getIntOrNull(COMPACTION_JOB_CREATION_LAMBDA_CONCURRENCY_MAXIMUM))
+                .reportBatchItemFailures(true)
                 .build());
 
         jobCreationQueue.grantSendMessages(triggerFunction);

java/deployment/cdk/src/main/java/sleeper/cdk/stack/core/TransactionLogSnapshotStack.java

@@ -135,7 +135,9 @@ private void createSnapshotCreationLambda(SleeperInstanceProps props, SleeperLam
 
         snapshotCreationLambda.addEventSource(SqsEventSource.Builder.create(queue)
                 .batchSize(instanceProperties.getInt(SNAPSHOT_CREATION_BATCH_SIZE))
-                .maxConcurrency(instanceProperties.getIntOrNull(SNAPSHOT_CREATION_LAMBDA_CONCURRENCY_MAXIMUM)).build());
+                .maxConcurrency(instanceProperties.getIntOrNull(SNAPSHOT_CREATION_LAMBDA_CONCURRENCY_MAXIMUM))
+                .reportBatchItemFailures(true)
+                .build());
 
         coreStacks.grantReadTablesStatus(snapshotCreationTrigger);
         coreStacks.grantInvokeScheduled(snapshotCreationTrigger, queue);
@@ -200,7 +202,9 @@ private void createSnapshotDeletionLambda(SleeperInstanceProps props, SleeperLam
 
         snapshotDeletionLambda.addEventSource(SqsEventSource.Builder.create(queue)
                 .batchSize(instanceProperties.getInt(SNAPSHOT_DELETION_BATCH_SIZE))
-                .maxConcurrency(instanceProperties.getIntOrNull(SNAPSHOT_DELETION_LAMBDA_CONCURRENCY_MAXIMUM)).build());
+                .maxConcurrency(instanceProperties.getIntOrNull(SNAPSHOT_DELETION_LAMBDA_CONCURRENCY_MAXIMUM))
+                .reportBatchItemFailures(true)
+                .build());
 
         coreStacks.grantReadTablesStatus(snapshotDeletionTrigger);
         coreStacks.grantInvokeScheduled(snapshotDeletionTrigger, queue);

java/deployment/cdk/src/main/java/sleeper/cdk/stack/core/TransactionLogTransactionStack.java

@@ -134,7 +134,9 @@ private void createTransactionDeletionLambda(SleeperInstanceProps props, Sleeper
 
         transactionDeletionLambda.addEventSource(SqsEventSource.Builder.create(queue)
                 .batchSize(instanceProperties.getInt(TRANSACTION_DELETION_BATCH_SIZE))
-                .maxConcurrency(instanceProperties.getIntOrNull(TRANSACTION_DELETION_LAMBDA_CONCURRENCY_MAXIMUM)).build());
+                .maxConcurrency(instanceProperties.getIntOrNull(TRANSACTION_DELETION_LAMBDA_CONCURRENCY_MAXIMUM))
+                .reportBatchItemFailures(true)
+                .build());
 
         coreStacks.grantReadTablesStatus(transactionDeletionTrigger);
         coreStacks.grantInvokeScheduled(transactionDeletionTrigger, queue);

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

@@ -9658,6 +9658,9 @@
                 "Arn"
               ]
             },
+            "FunctionResponseTypes": [
+              "ReportBatchItemFailures"
+            ],
             "Tags": [
               {
                 "Key": "DeploymentStack",
@@ -10735,6 +10738,9 @@
                 "Arn"
               ]
             },
+            "FunctionResponseTypes": [
+              "ReportBatchItemFailures"
+            ],
             "Tags": [
               {
                 "Key": "DeploymentStack",
@@ -11387,6 +11393,9 @@
                 "Arn"
               ]
             },
+            "FunctionResponseTypes": [
+              "ReportBatchItemFailures"
+            ],
             "Tags": [
               {
                 "Key": "DeploymentStack",
@@ -12785,6 +12794,9 @@
                 "Arn"
               ]
             },
+            "FunctionResponseTypes": [
+              "ReportBatchItemFailures"
+            ],
             "Tags": [
               {
                 "Key": "DeploymentStack",
@@ -16802,6 +16814,9 @@
                 "Arn"
               ]
             },
+            "FunctionResponseTypes": [
+              "ReportBatchItemFailures"
+            ],
             "Tags": [
               {
                 "Key": "DeploymentStack",
@@ -19979,6 +19994,9 @@
                 "Arn"
               ]
             },
+            "FunctionResponseTypes": [
+              "ReportBatchItemFailures"
+            ],
             "Tags": [
               {
                 "Key": "DeploymentStack",