fix(test, workflow-operator): stop leaking S3 test objects

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) ""

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