From 92e8e52ba72a7bb6c7369ce867db0535d2b67c57 Mon Sep 17 00:00:00 2001 From: Shawn Garello Date: Tue, 13 Oct 2026 22:59:35 -0700 Subject: [PATCH 1/2] fix(test, workflow-operator): stop leaking S3 test objects The LARGE_BINARY read test streams a real object to S3 whenever an endpoint is reachable and never deleted it, so every local run with the dev stack up left one more object under objects/8888/. Delete the test execution's objects in the existing outer finally, and assert at the end of the test that none are left. Use a negative execution id (-8670): real execution ids are positive, so the prefix delete can never remove a real run's binaries. Closes #8670 Generated-by: Claude Code (Opus 5.5) --- .../scan/file/FileScanSourceOpExecSpec.scala | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExecSpec.scala index 36c8ffb4e92..ef20a5f41ef 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExecSpec.scala @@ -38,8 +38,11 @@ import java.util.zip.{ZipEntry, ZipOutputStream} */ class FileScanSourceOpExecSpec extends AnyFlatSpec with BeforeAndAfterAll { - /** Execution id whose object prefix the large binaries created here are written under. */ - private val testExecutionId: Long = 8888L + /** + * Execution id whose object prefix the large binaries created here are written under. + * Negative because real execution ids are positive, so deleting its prefix spares real runs. + */ + private val testExecutionId: Long = -8670L private val testDir = Path .of(sys.env.getOrElse("TEXERA_HOME", ".")) @@ -175,6 +178,9 @@ class FileScanSourceOpExecSpec extends AnyFlatSpec with BeforeAndAfterAll { // An empty value clears the thread-local, so the seeded base URI cannot leak into // another suite that happens to reuse this thread. LargeBinaryManager.setCurrentBaseUri("") + // Delete what the scan uploaded even when it throws, or every run leaves an object + // behind. deleteByExecution logs failures instead of throwing, so it cannot displace one. + LargeBinaryManager.deleteByExecution(testExecutionId) } assert(tuples.size == 1) @@ -185,6 +191,11 @@ class FileScanSourceOpExecSpec extends AnyFlatSpec with BeforeAndAfterAll { assert(field.isInstanceOf[LargeBinary]) assert(field.asInstanceOf[LargeBinary].getUri.startsWith("s3://")) + + // Outside CI the scan uploads into whatever S3 the developer's config points at, so nothing + // it wrote may outlive the test. + val prefix = s"objects/$testExecutionId/" + assert(!S3StorageClient.directoryExists(LargeBinaryManager.DEFAULT_BUCKET, prefix), prefix) } // LargeBinary Tests From 47f56654ba38c5a00899d3ef1bb415a3e961d9de Mon Sep 17 00:00:00 2001 From: Shawn Garello Date: Thu, 15 Oct 2026 16:57:12 -0700 Subject: [PATCH 2/2] fix(test, workflow-operator): tie the large binary to the cleaned prefix Address review: assert that the produced object's URI is under the test's prefix, so the write side and the cleanup can't drift apart, and make the comments describe the current behaviour. Update baseUriForExecution's doc to allow negative test ids. Generated-by: Claude Code (Opus 5.5) --- .../texera/service/util/LargeBinaryManager.scala | 7 ++++--- .../source/scan/file/FileScanSourceOpExecSpec.scala | 12 ++++++------ 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/common/workflow-core/src/main/scala/org/apache/texera/service/util/LargeBinaryManager.scala b/common/workflow-core/src/main/scala/org/apache/texera/service/util/LargeBinaryManager.scala index 3ccd281db80..876e15a4940 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/service/util/LargeBinaryManager.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/service/util/LargeBinaryManager.scala @@ -36,9 +36,10 @@ object LargeBinaryManager extends LazyLogging { * Base URI (trailing slash) under which `executionId`'s large binaries live; create() * appends a unique suffix. Empty when the bucket is unconfigured, so create() fails loudly. * - * `executionId` must be a persisted EID. The sentinel id (DEFAULT_EXECUTION_ID = 1) shares - * this space, so binaries must only be created under a real execution — else execution 1 - * and a default-context run would collide under objects/1/. + * In production `executionId` is a persisted EID. The sentinel id (DEFAULT_EXECUTION_ID = 1) + * shares this space, so binaries must only be created under a real execution — else + * execution 1 and a default-context run would collide under objects/1/. Tests may use a + * negative id, which no real execution has. */ def baseUriForExecution(executionId: Long): String = if (DEFAULT_BUCKET.isEmpty) "" diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExecSpec.scala index ef20a5f41ef..60bcf1307e0 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExecSpec.scala @@ -39,8 +39,8 @@ import java.util.zip.{ZipEntry, ZipOutputStream} class FileScanSourceOpExecSpec extends AnyFlatSpec with BeforeAndAfterAll { /** - * Execution id whose object prefix the large binaries created here are written under. - * Negative because real execution ids are positive, so deleting its prefix spares real runs. + * Execution id the large binaries created here are written under. Real execution ids are + * positive, so a negative one gives this test an object prefix of its own. */ private val testExecutionId: Long = -8670L @@ -178,8 +178,8 @@ class FileScanSourceOpExecSpec extends AnyFlatSpec with BeforeAndAfterAll { // An empty value clears the thread-local, so the seeded base URI cannot leak into // another suite that happens to reuse this thread. LargeBinaryManager.setCurrentBaseUri("") - // Delete what the scan uploaded even when it throws, or every run leaves an object - // behind. deleteByExecution logs failures instead of throwing, so it cannot displace one. + // Deletes everything the scan uploaded under the test's prefix, also when it throws. + // deleteByExecution logs failures instead of throwing, so the test's own failure surfaces. LargeBinaryManager.deleteByExecution(testExecutionId) } @@ -191,9 +191,9 @@ class FileScanSourceOpExecSpec extends AnyFlatSpec with BeforeAndAfterAll { assert(field.isInstanceOf[LargeBinary]) assert(field.asInstanceOf[LargeBinary].getUri.startsWith("s3://")) + assert(field.asInstanceOf[LargeBinary].getUri.contains(s"objects/$testExecutionId/")) - // Outside CI the scan uploads into whatever S3 the developer's config points at, so nothing - // it wrote may outlive the test. + // The scan wrote under the test's prefix, and the cleanup above left nothing there. val prefix = s"objects/$testExecutionId/" assert(!S3StorageClient.directoryExists(LargeBinaryManager.DEFAULT_BUCKET, prefix), prefix) }