Issue 7564 - Configure bulk import S3 file retention

docs/usage/properties/instance/user/bulk_import.md

@@ -11,6 +11,7 @@ Note that on EMR, the total resource allocation must align with the instance typ
 | sleeper.bulk.import.emr.spark.speculation                                    | If true then speculative execution of tasks will be performed. Used to set spark.speculation.<br>See https://spark.apache.org/docs/latest/configuration.html.                                                                                                                                                                                                                                                                                                                                                                             | false                                                                                                                                                                                                                                                                                                           | true                        |
 | sleeper.bulk.import.spark.speculation.quantile                               | Fraction of tasks which must be complete before speculation is enabled for a particular stage. Used to set spark.speculation.quantile.<br>See https://spark.apache.org/docs/latest/configuration.html.                                                                                                                                                                                                                                                                                                                                    | 0.75                                                                                                                                                                                                                                                                                                            | true                        |
 | sleeper.bulk.import.starter.memory.mb                                        | The amount of memory in MB for lambda functions that start bulk import jobs.                                                                                                                                                                                                                                                                                                                                                                                                                                                              |                                                                                                                                                                                                                                                                                                                 | true                        |
+| sleeper.bulk.import.job.file.retention.days                                  | The number of days a bulk import job is held in the bulk import bucket. When a job is submitted to Spark it is written to S3 for Spark to read. This is retained in case the job needs to be retried, and for diagnostic purposes.<br>Defaults to the value of sleeper.log.retention.days.                                                                                                                                                                                                                                                |                                                                                                                                                                                                                                                                                                                 | true                        |
 | sleeper.bulk.import.emr.spark.executor.memory                                | The amount of memory allocated to a Spark executor. Used to set spark.executor.memory.<br>See https://spark.apache.org/docs/latest/configuration.html.                                                                                                                                                                                                                                                                                                                                                                                    | 16g                                                                                                                                                                                                                                                                                                             | true                        |
 | sleeper.bulk.import.emr.spark.driver.memory                                  | The amount of memory allocated to the Spark driver. Used to set spark.driver.memory.<br>See https://spark.apache.org/docs/latest/configuration.html.                                                                                                                                                                                                                                                                                                                                                                                      | 16g                                                                                                                                                                                                                                                                                                             | true                        |
 | sleeper.bulk.import.emr.spark.executor.instances                             | The number of executors. Used to set spark.executor.instances.<br>See https://spark.apache.org/docs/latest/configuration.html.                                                                                                                                                                                                                                                                                                                                                                                                            | 29                                                                                                                                                                                                                                                                                                              | true                        |

example/full/instance.properties

@@ -677,6 +677,13 @@
 # (default value shown below, uncomment to set a value)
 # sleeper.bulk.import.starter.memory.mb=4096
 
+# The number of days a bulk import job is held in the bulk import bucket. When a job is submitted to
+# Spark it is written to S3 for Spark to read. This is retained in case the job needs to be retried,
+# and for diagnostic purposes.
+# Defaults to the value of sleeper.log.retention.days.
+# (default value shown below, uncomment to set a value)
+# sleeper.bulk.import.job.file.retention.days=30
+
 # The amount of memory allocated to a Spark executor. Used to set spark.executor.memory.
 # See https://spark.apache.org/docs/latest/configuration.html.
 # (default value shown below, uncomment to set a value)

java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java

@@ -26,6 +26,12 @@
  * POJO containing information needed to run a bulk import job.
  */
 public class BulkImportJob {
+
+    /**
+     * The prefix for files holding a bulk import job in the bulk import bucket.
+     */
+    public static final String FILES_BUCKET_PREFIX = "bulk_import/";
+
     private final String id;
     private final String tableName;
     private final String tableId;

java/bulk-import/bulk-import-runner/src/main/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3.java

@@ -15,11 +15,9 @@
  */
 package sleeper.bulkimport.runner;
 
-import com.google.gson.JsonSyntaxException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.s3.S3Client;
-import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
 
 import sleeper.bulkimport.core.job.BulkImportJob;
@@ -46,16 +44,6 @@ public static BulkImportJob loadJob(InstanceProperties instanceProperties, Strin
                 .bucket(bulkImportBucket)
                 .key(objectKey)
                 .build()).asUtf8String();
-        try {
-            return new BulkImportJobSerDe().fromJson(jsonJob);
-        } catch (JsonSyntaxException e) {
-            LOGGER.error("Json job was malformed");
-            throw e;
-        } finally {
-            s3Client.deleteObject(DeleteObjectRequest.builder()
-                    .bucket(bulkImportBucket)
-                    .key(objectKey)
-                    .build());
-        }
+        return new BulkImportJobSerDe().fromJson(jsonJob);
     }
 }

java/bulk-import/bulk-import-runner/src/test/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3IT.java

@@ -51,13 +51,21 @@ void shouldLoadBulkImportJobFromS3() {
                 .files(List.of("/load-job.parquet"))
                 .build();
 
-        BulkImportJobWriterToS3 bulkImportJobWriterToS3 = new BulkImportJobWriterToS3(instanceProperties, s3Client);
-        bulkImportJobWriterToS3.writeJobToBulkImportBucket(bulkImportJob, objectKey);
-
-        // When / Then
-        assertThat(BulkImportJobLoaderFromS3.loadJob(instanceProperties, objectKey, s3Client))
-                .isEqualTo(bulkImportJob);
-        // And the file is deleted after it is loaded
-        assertThat(listObjectKeys(instanceProperties.get(BULK_IMPORT_BUCKET))).isEmpty();
+        // When
+        writer().writeJobToBulkImportBucket(bulkImportJob, objectKey);
+        BulkImportJob foundJob = loadJob(objectKey);
+
+        // Then
+        assertThat(foundJob).isEqualTo(bulkImportJob);
+        // And the file is kept
+        assertThat(listObjectKeys(instanceProperties.get(BULK_IMPORT_BUCKET))).containsExactly(objectKey);
+    }
+
+    private BulkImportJobWriterToS3 writer() {
+        return new BulkImportJobWriterToS3(instanceProperties, s3Client);
+    }
+
+    private BulkImportJob loadJob(String objectKey) {
+        return BulkImportJobLoaderFromS3.loadJob(instanceProperties, objectKey, s3Client);
     }
 }

java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportExecutor.java

@@ -58,7 +58,7 @@ public BulkImportExecutor(
     }
 
     public static String createJobFileObjectKey(BulkImportJob job, String jobRunId) {
-        return "bulk_import/" + job.getId() + "-" + jobRunId + ".json";
+        return BulkImportJob.FILES_BUCKET_PREFIX + job.getId() + "-" + jobRunId + ".json";
     }
 
     public void runJob(BulkImportJob bulkImportJob) {

java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportJobWriterToS3.java

@@ -30,8 +30,8 @@
 public class BulkImportJobWriterToS3 implements BulkImportExecutor.WriteJobToBucket {
     public static final Logger LOGGER = LoggerFactory.getLogger(BulkImportJobWriterToS3.class);
 
-    protected final InstanceProperties instanceProperties;
-    protected final S3Client s3Client;
+    private final InstanceProperties instanceProperties;
+    private final S3Client s3Client;
 
     public BulkImportJobWriterToS3(InstanceProperties instanceProperties, S3Client s3Client) {
         this.instanceProperties = instanceProperties;

java/core/src/main/java/sleeper/core/properties/instance/BulkImportProperty.java

@@ -21,6 +21,7 @@
 
 import java.util.List;
 
+import static sleeper.core.properties.instance.CommonProperty.LOG_RETENTION_IN_DAYS;
 import static sleeper.core.properties.instance.TableStateProperty.DEFAULT_TABLE_STATE_LAMBDA_MEMORY;
 
 /**
@@ -57,6 +58,15 @@ public interface BulkImportProperty {
             .defaultProperty(DEFAULT_TABLE_STATE_LAMBDA_MEMORY)
             .propertyGroup(InstancePropertyGroup.BULK_IMPORT)
             .runCdkDeployWhenChanged(true).build();
+    UserDefinedInstanceProperty BULK_IMPORT_JOB_FILE_RETENTION_DAYS = Index.propertyBuilder("sleeper.bulk.import.job.file.retention.days")
+            .description("The number of days a bulk import job is held in the bulk import bucket. When a job is " +
+                    "submitted to Spark it is written to S3 for Spark to read. This is retained in case the job " +
+                    "needs to be retried, and for diagnostic purposes.\n" +
+                    "Defaults to the value of sleeper.log.retention.days.")
+            .defaultProperty(LOG_RETENTION_IN_DAYS)
+            .validationPredicate(SleeperPropertyValueUtils::isPositiveInteger)
+            .propertyGroup(InstancePropertyGroup.BULK_IMPORT)
+            .runCdkDeployWhenChanged(true).build();
 
     static List<UserDefinedInstanceProperty> getAll() {
         return Index.INSTANCE.getAll();

java/core/src/test/java/sleeper/core/properties/SleeperPropertiesValidationTest.java

@@ -29,10 +29,10 @@
 import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static sleeper.core.properties.PropertiesUtils.loadProperties;
+import static sleeper.core.properties.instance.BulkImportProperty.BULK_IMPORT_JOB_FILE_RETENTION_DAYS;
 import static sleeper.core.properties.instance.CdkDefinedInstanceProperty.VERSION;
 import static sleeper.core.properties.instance.CommonProperty.ID;
 import static sleeper.core.properties.instance.CommonProperty.JARS_BUCKET;
-import static sleeper.core.properties.instance.CommonProperty.LOG_RETENTION_IN_DAYS;
 import static sleeper.core.properties.instance.CommonProperty.MAXIMUM_CONNECTIONS_TO_S3;
 import static sleeper.core.properties.instance.CommonProperty.SUBNETS;
 import static sleeper.core.properties.instance.CommonProperty.VPC_ID;
@@ -168,17 +168,17 @@ class MultipleValidationErrors {
         void shouldFailValidationWithTwoInvalidInstanceProperties() {
             // Given
             InstanceProperties instanceProperties = createTestInstanceProperties();
-            instanceProperties.set(LOG_RETENTION_IN_DAYS, "abc");
+            instanceProperties.set(BULK_IMPORT_JOB_FILE_RETENTION_DAYS, "abc");
             instanceProperties.set(MAXIMUM_CONNECTIONS_TO_S3, "def");
 
             // When
             assertThatThrownBy(instanceProperties::validate)
                     .isInstanceOf(SleeperPropertiesInvalidException.class)
-                    .hasMessageContaining("Property sleeper.log.retention.days was invalid. It was \"abc\".")
+                    .hasMessageContaining("Property sleeper.fs.s3a.max-connections was invalid. It was \"def\".")
                     .hasMessageContaining("Failure 1 of 2.")
                     .extracting("invalidValues")
                     .isEqualTo(Map.of(
-                            LOG_RETENTION_IN_DAYS, "abc",
+                            BULK_IMPORT_JOB_FILE_RETENTION_DAYS, "abc",
                             MAXIMUM_CONNECTIONS_TO_S3, "def"));
         }
 

java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java

@@ -15,19 +15,26 @@
  */
 package sleeper.cdk.stack.bulkimport;
 
+import software.amazon.awscdk.Duration;
 import software.amazon.awscdk.NestedStack;
 import software.amazon.awscdk.RemovalPolicy;
 import software.amazon.awscdk.services.s3.BlockPublicAccess;
 import software.amazon.awscdk.services.s3.Bucket;
 import software.amazon.awscdk.services.s3.BucketEncryption;
 import software.amazon.awscdk.services.s3.IBucket;
+import software.amazon.awscdk.services.s3.LifecycleRule;
 import software.constructs.Construct;
 
+import sleeper.bulkimport.core.job.BulkImportJob;
 import sleeper.cdk.stack.SleeperCoreStacks;
 import sleeper.cdk.util.S3BucketName;
 import sleeper.core.properties.instance.InstanceProperties;
 
+import java.util.List;
+
+import static sleeper.core.properties.instance.BulkImportProperty.BULK_IMPORT_JOB_FILE_RETENTION_DAYS;
 import static sleeper.core.properties.instance.CdkDefinedInstanceProperty.BULK_IMPORT_BUCKET;
+import static sleeper.core.properties.instance.CommonProperty.LOG_RETENTION_IN_DAYS;
 
 public class BulkImportBucketStack extends NestedStack {
     private final IBucket importBucket;
@@ -42,6 +49,19 @@ public BulkImportBucketStack(Construct scope, String id, InstanceProperties inst
                 .versioned(false)
                 .removalPolicy(RemovalPolicy.DESTROY)
                 .encryption(BucketEncryption.S3_MANAGED)
+                .lifecycleRules(List.of(
+                        LifecycleRule.builder()
+                                .prefix(BulkImportJob.FILES_BUCKET_PREFIX)
+                                .expiration(Duration.days(instanceProperties.getInt(BULK_IMPORT_JOB_FILE_RETENTION_DAYS)))
+                                .build(),
+                        LifecycleRule.builder()
+                                .prefix("logs/")
+                                .expiration(Duration.days(instanceProperties.getInt(LOG_RETENTION_IN_DAYS)))
+                                .build(),
+                        LifecycleRule.builder()
+                                .prefix("applications/")
+                                .expiration(Duration.days(instanceProperties.getInt(LOG_RETENTION_IN_DAYS)))
+                                .build()))
                 .build();
         importBucket.grantWrite(coreStacks.getIngestByQueuePolicyForGrants());
         instanceProperties.set(BULK_IMPORT_BUCKET, importBucket.getBucketName());

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

@@ -13210,6 +13210,25 @@
                   }
                 }
               ]
+            },
+            "LifecycleConfiguration": {
+              "Rules": [
+                {
+                  "ExpirationInDays": 30,
+                  "Prefix": "bulk_import/",
+                  "Status": "Enabled"
+                },
+                {
+                  "ExpirationInDays": 30,
+                  "Prefix": "logs/",
+                  "Status": "Enabled"
+                },
+                {
+                  "ExpirationInDays": 30,
+                  "Prefix": "applications/",
+                  "Status": "Enabled"
+                }
+              ]
             }
           }
         },

scripts/templates/instanceproperties.template

@@ -681,6 +681,13 @@
 # (default value shown below, uncomment to set a value)
 # sleeper.bulk.import.starter.memory.mb=4096
 
+# The number of days a bulk import job is held in the bulk import bucket. When a job is submitted to
+# Spark it is written to S3 for Spark to read. This is retained in case the job needs to be retried,
+# and for diagnostic purposes.
+# Defaults to the value of sleeper.log.retention.days.
+# (default value shown below, uncomment to set a value)
+# sleeper.bulk.import.job.file.retention.days=30
+
 # The amount of memory allocated to a Spark executor. Used to set spark.executor.memory.
 # See https://spark.apache.org/docs/latest/configuration.html.
 # (default value shown below, uncomment to set a value)