From 5aa6cedbdfc99c09c22effb6b710c50e4605c8a4 Mon Sep 17 00:00:00 2001 From: patchwork01 <110390516+patchwork01@users.noreply.github.com> Date: Mon, 20 Jul 2026 11:04:47 +0000 Subject: [PATCH 1/6] 7564 Stop deleting bulk import job file --- .../runner/BulkImportJobLoaderFromS3.java | 14 +---------- .../runner/BulkImportJobLoaderFromS3IT.java | 24 ++++++++++++------- .../executor/BulkImportJobWriterToS3.java | 4 ++-- 3 files changed, 19 insertions(+), 23 deletions(-) diff --git a/java/bulk-import/bulk-import-runner/src/main/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3.java b/java/bulk-import/bulk-import-runner/src/main/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3.java index 90036409261..70b417f9a2a 100644 --- a/java/bulk-import/bulk-import-runner/src/main/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3.java +++ b/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); } } diff --git a/java/bulk-import/bulk-import-runner/src/test/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3IT.java b/java/bulk-import/bulk-import-runner/src/test/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3IT.java index fa83195c4f1..53ad4575c43 100644 --- a/java/bulk-import/bulk-import-runner/src/test/java/sleeper/bulkimport/runner/BulkImportJobLoaderFromS3IT.java +++ b/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); } } diff --git a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportJobWriterToS3.java b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportJobWriterToS3.java index 4cef05283bb..a0daaa8a95a 100644 --- a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportJobWriterToS3.java +++ b/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; From b3c8b6029caabc1c45ab4c010055a4e780193d22 Mon Sep 17 00:00:00 2001 From: patchwork01 <110390516+patchwork01@users.noreply.github.com> Date: Mon, 20 Jul 2026 11:36:16 +0000 Subject: [PATCH 2/6] 7564 Configure lifecycle rules for bulk import bucket --- .../properties/instance/user/bulk_import.md | 1 + example/full/instance.properties | 7 +++++++ .../bulkimport/core/job/BulkImportJob.java | 6 ++++++ .../starter/executor/BulkImportExecutor.java | 2 +- .../properties/instance/BulkImportProperty.java | 10 ++++++++++ .../stack/bulkimport/BulkImportBucketStack.java | 16 ++++++++++++++++ .../sleeper/cdk/default-instance.approved.json | 14 ++++++++++++++ scripts/templates/instanceproperties.template | 7 +++++++ 8 files changed, 62 insertions(+), 1 deletion(-) diff --git a/docs/usage/properties/instance/user/bulk_import.md b/docs/usage/properties/instance/user/bulk_import.md index 98c7d4ea788..0ac33cb8085 100644 --- a/docs/usage/properties/instance/user/bulk_import.md +++ b/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.
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.
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.
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.
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.
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.
See https://spark.apache.org/docs/latest/configuration.html. | 29 | true | diff --git a/example/full/instance.properties b/example/full/instance.properties index 84480d7d91e..459f947b238 100644 --- a/example/full/instance.properties +++ b/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) diff --git a/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java b/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java index 6819a0afd4e..52af825c6a0 100644 --- a/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java +++ b/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 String FILES_BUCKET_PREFIX = "bulk_import/"; + private final String id; private final String tableName; private final String tableId; diff --git a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportExecutor.java b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportExecutor.java index 3426a2eb13f..9cdf971b1c0 100644 --- a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportExecutor.java +++ b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/BulkImportExecutor.java @@ -75,7 +75,7 @@ public void runJob(BulkImportJob bulkImportJob, String jobRunId) { .jobRunId(jobRunId).build()); try { LOGGER.info("Writing job with id {} to JSON file", bulkImportJob.getId()); - String jobFileObjectKey = "bulk_import/" + bulkImportJob.getId() + "-" + jobRunId + ".json"; + String jobFileObjectKey = BulkImportJob.FILES_BUCKET_PREFIX + bulkImportJob.getId() + "-" + jobRunId + ".json"; writeJobToBucket.writeJobToBulkImportBucket(bulkImportJob, jobFileObjectKey); LOGGER.info("Submitting job with id {}", bulkImportJob.getId()); platformExecutor.runJobOnPlatform(BulkImportArguments.builder() diff --git a/java/core/src/main/java/sleeper/core/properties/instance/BulkImportProperty.java b/java/core/src/main/java/sleeper/core/properties/instance/BulkImportProperty.java index bb2419a9337..81c8f636777 100644 --- a/java/core/src/main/java/sleeper/core/properties/instance/BulkImportProperty.java +++ b/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 getAll() { return Index.INSTANCE.getAll(); diff --git a/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java b/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java index 272785ec539..00b93d5c311 100644 --- a/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java +++ b/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,15 @@ 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())) .build(); importBucket.grantWrite(coreStacks.getIngestByQueuePolicyForGrants()); instanceProperties.set(BULK_IMPORT_BUCKET, importBucket.getBucketName()); diff --git a/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json b/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json index f07d4328db8..a79278f96a4 100644 --- a/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json +++ b/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json @@ -13201,6 +13201,20 @@ } } ] + }, + "LifecycleConfiguration": { + "Rules": [ + { + "ExpirationInDays": 30, + "Prefix": "bulk_import/", + "Status": "Enabled" + }, + { + "ExpirationInDays": 30, + "Prefix": "logs/", + "Status": "Enabled" + } + ] } } }, diff --git a/scripts/templates/instanceproperties.template b/scripts/templates/instanceproperties.template index d8bc4798ad2..39d0372f9de 100644 --- a/scripts/templates/instanceproperties.template +++ b/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) From 6d0d4c646c83c4c6527235ccfc978a3f8bee8b2e Mon Sep 17 00:00:00 2001 From: patchwork01 <110390516+patchwork01@users.noreply.github.com> Date: Mon, 20 Jul 2026 11:50:19 +0000 Subject: [PATCH 3/6] 7564 Add missing final keyword --- .../main/java/sleeper/bulkimport/core/job/BulkImportJob.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java b/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java index 52af825c6a0..4b9395fa669 100644 --- a/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java +++ b/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/job/BulkImportJob.java @@ -30,7 +30,7 @@ public class BulkImportJob { /** * The prefix for files holding a bulk import job in the bulk import bucket. */ - public static String FILES_BUCKET_PREFIX = "bulk_import/"; + public static final String FILES_BUCKET_PREFIX = "bulk_import/"; private final String id; private final String tableName; From d7c0bbfe18d666f2536b9b7afe41ec85c997235e Mon Sep 17 00:00:00 2001 From: patchwork01 <110390516+patchwork01@users.noreply.github.com> Date: Mon, 20 Jul 2026 11:53:18 +0000 Subject: [PATCH 4/6] 7564 Fix SleeperPropertiesValidationTest --- .../core/properties/SleeperPropertiesValidationTest.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/java/core/src/test/java/sleeper/core/properties/SleeperPropertiesValidationTest.java b/java/core/src/test/java/sleeper/core/properties/SleeperPropertiesValidationTest.java index 88d142e8674..62d55acaf24 100644 --- a/java/core/src/test/java/sleeper/core/properties/SleeperPropertiesValidationTest.java +++ b/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")); } From 4fea8eb474c876cfa36427123f8848b465f913a0 Mon Sep 17 00:00:00 2001 From: patchwork01 <110390516+patchwork01@users.noreply.github.com> Date: Mon, 20 Jul 2026 13:18:10 +0000 Subject: [PATCH 5/6] 7564 Lifecycle rule for bulk import applications folder --- .../sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java b/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java index 00b93d5c311..c02085c3186 100644 --- a/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java +++ b/java/deployment/cdk/src/main/java/sleeper/cdk/stack/bulkimport/BulkImportBucketStack.java @@ -57,6 +57,10 @@ public BulkImportBucketStack(Construct scope, String id, InstanceProperties inst 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()); From 5515a7730c039e170287aed344b02d5dfd9995c1 Mon Sep 17 00:00:00 2001 From: patchwork01 <110390516+patchwork01@users.noreply.github.com> Date: Mon, 20 Jul 2026 13:42:52 +0000 Subject: [PATCH 6/6] 7564 Fix SleeperInstanceIT --- .../src/test/java/sleeper/cdk/default-instance.approved.json | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json b/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json index 4cf5d531426..3ef54c8c8ab 100644 --- a/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json +++ b/java/deployment/cdk/src/test/java/sleeper/cdk/default-instance.approved.json @@ -13241,6 +13241,11 @@ "ExpirationInDays": 30, "Prefix": "logs/", "Status": "Enabled" + }, + { + "ExpirationInDays": 30, + "Prefix": "applications/", + "Status": "Enabled" } ] }