Issue 2379 - Call FIFO queue from GC client

.github/config/chunks.yaml

@@ -18,7 +18,7 @@ chunks:
   common:
     name: Common
     workflow: chunk-common.yaml
-    modules: [ core, configuration, sketches, parquet, common/common-job, common/common-task, build, common/dynamodb-tools ]
+    modules: [ core, configuration, sketches, parquet, common/common-job, common/common-task, common/common-invoke-tables, build, common/dynamodb-tools ]
   compaction:
     name: Compaction
     workflow: chunk-compaction.yaml

.github/workflows/chunk-cdk.yaml

@@ -14,10 +14,10 @@ on:
       - 'java/cdk-environment/**'
       - 'java/system-test/system-test-cdk/**'
       - 'java/system-test/system-test-configuration/**'
+      - 'java/system-test/system-test-data-generation/**'
       - 'java/system-test/system-test-dsl/**'
       - 'java/system-test/system-test-drivers/**'
       - 'java/system-test/system-test-suite/**'
-      - 'java/system-test/system-test-data-generation/**'
       - 'java/bulk-import/bulk-import-starter/**'
       - 'java/clients/**'
       - 'java/ingest/ingest-batcher-submitter/**'
@@ -34,6 +34,7 @@ on:
       - 'java/ingest/ingest-runner/**'
       - 'java/compaction/compaction-status-store/**'
       - 'java/compaction/compaction-job-creation/**'
+      - 'java/common/common-invoke-tables/**'
       - 'java/common/common-job/**'
       - 'java/ingest/ingest-status-store/**'
       - 'java/sketches/**'

.github/workflows/chunk-clients.yaml

@@ -13,6 +13,7 @@ on:
       - 'java/compaction/compaction-status-store/**'
       - 'java/splitter/splitter-core/**'
       - 'java/ingest/ingest-batcher-store/**'
+      - 'java/common/common-invoke-tables/**'
       - 'java/common/common-task/**'
       - 'java/ingest/ingest-batcher-core/**'
       - 'java/compaction/compaction-core/**'

.github/workflows/chunk-common.yaml

@@ -15,6 +15,7 @@ on:
       - 'java/parquet/**'
       - 'java/common/common-job/**'
       - 'java/common/common-task/**'
+      - 'java/common/common-invoke-tables/**'
       - 'java/build/**'
       - 'java/common/dynamodb-tools/**'
 

.github/workflows/chunk-compaction.yaml

@@ -19,6 +19,7 @@ on:
       - 'java/splitter/splitter-core/**'
       - 'java/splitter/splitter-lambda/**'
       - 'java/common/common-task/**'
+      - 'java/common/common-invoke-tables/**'
       - 'java/ingest/ingest-runner/**'
       - 'java/common/common-job/**'
       - 'java/ingest/ingest-status-store/**'

.github/workflows/chunk-data.yaml

@@ -8,12 +8,13 @@ on:
       - 'code-style/checkstyle*.xml'
       - 'code-style/spotbugs*.xml'
       - 'java/pom.xml'
-      - 'java/metrics/**'
       - 'java/garbage-collector/**'
+      - 'java/metrics/**'
+      - 'java/common/common-invoke-tables/**'
       - 'java/statestore/**'
-      - 'java/common/dynamodb-tools/**'
       - 'java/parquet/**'
       - 'java/configuration/**'
+      - 'java/common/dynamodb-tools/**'
       - 'java/core/**'
 
 jobs:

java/clients/pom.xml

@@ -136,6 +136,11 @@
             <artifactId>ingest-batcher-store</artifactId>
             <version>${project.parent.version}</version>
         </dependency>
+        <dependency>
+            <groupId>sleeper</groupId>
+            <artifactId>common-invoke-tables</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
         <!-- Test dependencies -->
         <dependency>
             <groupId>sleeper</groupId>

java/clients/src/main/java/sleeper/clients/status/update/TriggerGarbageCollectionClient.java

@@ -24,11 +24,8 @@
 
 import sleeper.configuration.properties.instance.InstanceProperties;
 import sleeper.configuration.table.index.DynamoDBTableIndex;
-import sleeper.core.table.InvokeForTableRequest;
-import sleeper.core.table.InvokeForTableRequestSerDe;
 import sleeper.core.table.TableIndex;
-import sleeper.core.table.TableNotFoundException;
-import sleeper.core.table.TableStatus;
+import sleeper.invoke.tables.InvokeForTables;
 
 import java.util.List;
 import java.util.stream.Stream;
@@ -58,14 +55,8 @@ public static void main(String[] args) {
             InstanceProperties instanceProperties = new InstanceProperties();
             instanceProperties.loadFromS3GivenInstanceId(s3Client, instanceId);
             TableIndex tableIndex = new DynamoDBTableIndex(instanceProperties, dynamoClient);
-            List<TableStatus> tables = tableNames.stream()
-                    .map(name -> tableIndex.getTableByName(name)
-                            .orElseThrow(() -> TableNotFoundException.withTableName(name)))
-                    .collect(toUnmodifiableList());
             String queueUrl = instanceProperties.get(GARBAGE_COLLECTOR_QUEUE_URL);
-            InvokeForTableRequestSerDe serDe = new InvokeForTableRequestSerDe();
-            InvokeForTableRequest.forTables(tables.stream(), tables.size(),
-                    request -> sqsClient.sendMessage(queueUrl, serDe.toJson(request)));
+            InvokeForTables.sendOneMessagePerTableByName(sqsClient, queueUrl, tableIndex, tableNames);
         } finally {
             s3Client.shutdown();
             dynamoClient.shutdown();

java/common/common-invoke-tables/pom.xml

@@ -0,0 +1,75 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2022-2024 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>common</artifactId>
+        <groupId>sleeper</groupId>
+        <version>0.23.0-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>common-invoke-tables</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-sqs</artifactId>
+            <version>${aws-java-sdk.version}</version>
+        </dependency>
+        <!-- Sleeper dependencies -->
+        <dependency>
+            <groupId>sleeper</groupId>
+            <artifactId>core</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>
+        <dependency>
+            <groupId>sleeper</groupId>
+            <artifactId>configuration</artifactId>
+            <version>${project.parent.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>localstack</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

java/common/common-invoke-tables/src/main/java/sleeper/invoke/tables/InvokeForTables.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2022-2024 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.invoke.tables;
+
+import com.amazonaws.services.sqs.AmazonSQS;
+import com.amazonaws.services.sqs.model.SendMessageBatchRequest;
+import com.amazonaws.services.sqs.model.SendMessageBatchRequestEntry;
+
+import sleeper.core.table.TableIndex;
+import sleeper.core.table.TableNotFoundException;
+import sleeper.core.table.TableStatus;
+import sleeper.core.util.SplitIntoBatches;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Stream;
+
+import static java.util.stream.Collectors.toUnmodifiableList;
+
+public class InvokeForTables {
+
+    private InvokeForTables() {
+    }
+
+    public static void sendOneMessagePerTable(AmazonSQS sqsClient, String queueUrl, Stream<TableStatus> tables) {
+        SplitIntoBatches.reusingListOfSize(10, tables,
+                batch -> sendMessageBatch(sqsClient, queueUrl, batch));
+    }
+
+    public static void sendOneMessagePerTableByName(
+            AmazonSQS sqsClient, String queueUrl, TableIndex tableIndex, List<String> tableNames) {
+        List<TableStatus> tables = tableNames.stream().map(name -> tableIndex.getTableByName(name)
+                .orElseThrow(() -> TableNotFoundException.withTableName(name)))
+                .collect(toUnmodifiableList());
+        sendOneMessagePerTable(sqsClient, queueUrl, tables.stream());
+    }
+
+    private static void sendMessageBatch(AmazonSQS sqsClient, String queueUrl, List<TableStatus> tablesBatch) {
+        sqsClient.sendMessageBatch(new SendMessageBatchRequest()
+                .withQueueUrl(queueUrl)
+                .withEntries(tablesBatch.stream()
+                        .map(table -> new SendMessageBatchRequestEntry()
+                                .withMessageDeduplicationId(UUID.randomUUID().toString())
+                                .withId(table.getTableUniqueId())
+                                .withMessageGroupId(table.getTableUniqueId())
+                                .withMessageBody(table.getTableUniqueId()))
+                        .collect(toUnmodifiableList())));
+    }
+}