4472 - Reduce duplication code for compactions and bulk export

java/bulk-export/bulk-export-task-creator/src/main/java/sleeper/bulkexport/taskcreator/SqsTriggeredBulkExportTaskRunnerLambda.java

@@ -24,6 +24,7 @@
 import software.amazon.awssdk.services.sqs.SqsClient;
 
 import sleeper.common.task.QueueMessageCount;
+import sleeper.common.task.RunDataProcessingTasks;
 import sleeper.configuration.properties.S3InstanceProperties;
 import sleeper.core.properties.instance.InstanceProperties;
 
@@ -38,16 +39,17 @@ public class SqsTriggeredBulkExportTaskRunnerLambda {
     private static final Logger LOGGER = LoggerFactory.getLogger(SqsTriggeredBulkExportTaskRunnerLambda.class);
 
     private final QueueMessageCount.Client queueMessageCount;
-    private final RunLeafPartitionBulkExportTasks runLeafPartitionBulkExportTasks;
+    private final RunDataProcessingTasks runTasks;
 
     public SqsTriggeredBulkExportTaskRunnerLambda() {
         String s3Bucket = validateParameter(CONFIG_BUCKET.toEnvironmentVariable());
         SqsClient sqsClient = SqsClient.create();
         S3Client s3Client = S3Client.create();
         EcsClient ecsClient = EcsClient.create();
         InstanceProperties instanceProperties = S3InstanceProperties.loadFromBucket(s3Client, s3Bucket);
-        runLeafPartitionBulkExportTasks = new RunLeafPartitionBulkExportTasks(instanceProperties, ecsClient);
-        queueMessageCount = QueueMessageCount.withSqsClient(sqsClient);
+
+        this.runTasks = RunDataProcessingTasks.createForBulkExport(instanceProperties, ecsClient);
+        this.queueMessageCount = QueueMessageCount.withSqsClient(sqsClient);
     }
 
     /**
@@ -58,7 +60,7 @@ public SqsTriggeredBulkExportTaskRunnerLambda() {
      * @param context the lambda context
      */
     public void handleRequest(SQSEvent input, Context context) {
-        runLeafPartitionBulkExportTasks.run(queueMessageCount);
+        runTasks.run(queueMessageCount);
     }
 
     private static String validateParameter(String parameterName) {

java/common/common-task/src/main/java/sleeper/common/task/BulkExportTaskHostScaler.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2022-2025 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.common.task;
+
+public class BulkExportTaskHostScaler implements TaskHostScaler {
+
+    @Override
+    public void scaleTo(int numberOfTasks) {
+        // Bulk Export doesn't currently scale so do nothing
+    }
+
+}

java/common/common-task/src/main/java/sleeper/common/task/CompactionTaskHostScaler.java

@@ -37,7 +37,7 @@
  * Autoscaler to scale EC2 instances for the desired number of compaction tasks. This makes decisions on how many
  * instances to start and stop based on the amount of work there is to do.
  */
-public class CompactionTaskHostScaler {
+public class CompactionTaskHostScaler implements TaskHostScaler {
 
     private final InstanceProperties instanceProperties;
     private final CheckAutoScalingGroup asgQuery;
@@ -63,6 +63,7 @@ public CompactionTaskHostScaler(
      *
      * @param numberOfTasks total number of tasks to provide capacity for
      */
+    @Override
     public void scaleTo(int numberOfTasks) {
         String launchType = instanceProperties.get(COMPACTION_ECS_LAUNCHTYPE);
         // Only need scaler for EC2

java/common/common-task/src/main/java/sleeper/common/task/RunBulkExportTasks.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2022-2025 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.common.task;
+
+import software.amazon.awssdk.services.ecs.EcsClient;
+import software.amazon.awssdk.services.s3.S3Client;
+
+import sleeper.configuration.properties.S3InstanceProperties;
+import sleeper.core.properties.instance.InstanceProperties;
+
+/**
+ * Finds the number of messages on a queue, and starts up one EC2 or Fargate task for each, up to a
+ * configurable maximum.
+ */
+public class RunBulkExportTasks {
+
+    public static void main(String[] args) {
+        if (args.length != 2) {
+            System.out.println("Usage: <instance-id> <number-of-tasks>");
+            return;
+        }
+
+        try (S3Client s3Client = S3Client.create();
+                EcsClient ecsClient = EcsClient.create()) {
+            String instanceId = args[0];
+            int numberOfTasks = Integer.parseInt(args[1]);
+
+            InstanceProperties instanceProperties = S3InstanceProperties.loadGivenInstanceId(s3Client, instanceId);
+
+            RunDataProcessingTasks.createForBulkExport(instanceProperties, ecsClient)
+                    .runToMeetTargetTasks(numberOfTasks);
+        }
+    }
+
+    private RunBulkExportTasks() {
+    }
+}