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 36c8ffb4e92..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 @@ -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 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 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("") + // 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) } 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://")) + assert(field.asInstanceOf[LargeBinary].getUri.contains(s"objects/$testExecutionId/")) + + // 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) } // LargeBinary Tests