You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/07/29 04:51:25 UTC
(texera) branch gh-readonly-queue/main/pr-6987-ab58d477ba77d1f42cbe56f888dda96fe3fffad0 created (now f77e4cc5f3)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6987-ab58d477ba77d1f42cbe56f888dda96fe3fffad0
in repository https://gitbox.apache.org/repos/asf/texera.git
at f77e4cc5f3 test(workflow-core): extend util and tuple unit test coverage (#6987)
This branch includes the following new commits:
new f77e4cc5f3 test(workflow-core): extend util and tuple unit test coverage (#6987)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: test(workflow-core): extend util and tuple unit test coverage (#6987)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6987-ab58d477ba77d1f42cbe56f888dda96fe3fffad0
in repository https://gitbox.apache.org/repos/asf/texera.git
commit f77e4cc5f31cbe393397a333a25abd966b24cd0f
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Tue Jul 28 21:51:13 2026 -0700
test(workflow-core): extend util and tuple unit test coverage (#6987)
### What changes were proposed in this PR?
Adds 53 tests across four `common/workflow-core` classes (88 -> 141 in
these suites). No source changes.
- **`StackTraceUtils`** (was **0% covered**, new spec): pins the exact
rendered output (header + `Throwable.toString` + frames), the `Caused
by` recursion including a 20-deep chain, per-cause frames, and the
null-message case.
- **`AttributeTypeUtils`**: the unsupported-type `case _` branches of
`parseInteger`/`parseLong`/`parseDouble`/`parseBoolean` (asserting both
the message and the wrapped `IllegalArgumentException` cause), the
previously-unexercised Long/Integer/Timestamp/Boolean input arms,
forced-vs-strict divergence (forced `"2147483648"` wraps to
`Int.MinValue`; `"1.9"` truncates where strict throws), the `trim.toInt
== 1` fallback, schema-inference edges (empty, one-way widening,
all-null), and `tupleCasting` null handling.
- **`S3StorageClient`**: `uploadPartWithRequest`'s two payload branches
(known `contentLength` -> `fromInputStream`, unknown -> `readAllBytes`)
with a two-part ordered round trip, the empty-prefix `require` guard,
and the abort-on-mid-stream-failure path — asserting the exception
propagates, no pending upload id remains (proving `abortMultipartUpload`
ran), and no object was published. Follows the suite's existing
MinIO-testcontainer approach (no client double exists); no live AWS.
- **`LargeBinaryManager`**: `baseUriForExecution` formatting,
S3-verified re-seeding across executions, clear-on-empty/null making
`create()` fail loudly, and thread-local isolation.
Two branches were deliberately left uncovered rather than forced, and
are called out in the code review notes: the `DEFAULT_BUCKET.isEmpty`
arm (a val captured at object init — unreachable without mutating
production config) and a real per-key `DeleteObjects` failure (needs S3
object-lock setup; `throwOnDeleteErrors` is already tested directly).
### Any related issues, documentation, discussions?
Closes #6985.
### How was this PR tested?
`sbt -java-home <jbr-17> "WorkflowCore/testOnly *S3StorageClientSpec
*AttributeTypeUtilsSpec *StackTraceUtilsSpec *LargeBinaryManagerSpec"`
-> 141 succeeded, 0 failed (4 suites). `Test/scalafmtCheck` +
`Test/scalafix --check` clean.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
---
.../amber/core/tuple/AttributeTypeUtilsSpec.scala | 227 ++++++++++++++++
.../texera/amber/util/StackTraceUtilsSpec.scala | 143 +++++++++++
.../service/util/LargeBinaryManagerSpec.scala | 76 ++++++
.../texera/service/util/S3StorageClientSpec.scala | 284 ++++++++++++++++++++-
4 files changed, 727 insertions(+), 3 deletions(-)
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala
index b6ec7e5c79..38ba986256 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala
@@ -475,4 +475,231 @@ class AttributeTypeUtilsSpec extends AnyFunSuite {
}
assert(longEx.getMessage == "Failed to parse type java.lang.String to Long: abc")
}
+
+ test("parseField returns null for every target type when the field is null") {
+ // The null short-circuit runs before any per-type parsing.
+ Seq(INTEGER, LONG, DOUBLE, BOOLEAN, TIMESTAMP, STRING, BINARY, LARGE_BINARY, ANY).foreach {
+ attributeType => assert(parseField(null, attributeType) == null)
+ }
+ }
+
+ test("parseField to INTEGER accepts Long input and trims surrounding whitespace") {
+ assert(parseField(java.lang.Long.valueOf(42L), INTEGER) == 42)
+ // A Long beyond Int range is narrowed rather than rejected.
+ assert(parseField(java.lang.Long.valueOf(3000000000L), INTEGER) == 3000000000L.toInt)
+ assert(parseField(" 7\t\n", INTEGER) == 7)
+ assert(parseField(" 1,234 ", INTEGER, force = true) == 1234)
+ }
+
+ test("parseField to INTEGER rejects types that have no numeric meaning") {
+ // Timestamp / binary fall into the unsupported branch and surface as AttributeTypeException.
+ val timestampEx = intercept[AttributeTypeException] {
+ parseField(new Timestamp(0L), INTEGER)
+ }
+ assert(timestampEx.getMessage.startsWith("Failed to parse type java.sql.Timestamp to Integer:"))
+ assert(timestampEx.getCause.isInstanceOf[IllegalArgumentException])
+ assert(
+ timestampEx.getCause.getMessage ==
+ "Unsupported type for parsing to Integer: java.sql.Timestamp"
+ )
+
+ assertThrows[AttributeTypeException](parseField(Array[Byte](1, 2), INTEGER))
+ }
+
+ test("parseField to INTEGER overflows out of range values only under forced parsing") {
+ // Strict parsing rejects a value beyond Int range...
+ assertThrows[AttributeTypeException](parseField("2147483648", INTEGER))
+ // ...while the forced (locale-aware) path parses a Long and narrows it, wrapping around.
+ assert(parseField("2147483648", INTEGER, force = true) == Integer.MIN_VALUE)
+ // Forced parsing also accepts decimals and truncates toward zero, unlike strict parsing.
+ assertThrows[AttributeTypeException](parseField("1.9", INTEGER))
+ assert(parseField("1.9", INTEGER, force = true) == 1)
+ assert(parseField("-1.9", INTEGER, force = true) == -1)
+ }
+
+ test("parseField to LONG accepts Integer and Timestamp inputs") {
+ assert(parseField(java.lang.Integer.valueOf(42), LONG) == 42L)
+ assert(parseField(new Timestamp(1699820130000L), LONG) == 1699820130000L)
+ assert(parseField(false, LONG) == 0L)
+ assert(parseField(" -9223372036854775808 ", LONG) == Long.MinValue)
+ // Forced parsing truncates a decimal instead of failing.
+ assertThrows[AttributeTypeException](parseField("2.9", LONG))
+ assert(parseField("2.9", LONG, force = true) == 2L)
+ }
+
+ test("parseField to LONG rejects binary input") {
+ val ex = intercept[AttributeTypeException] {
+ parseField(Array[Byte](1, 2), LONG)
+ }
+ assert(ex.getCause.isInstanceOf[IllegalArgumentException])
+ assert(ex.getCause.getMessage.startsWith("Unsupported type for parsing to Long:"))
+ }
+
+ test("parseField to DOUBLE keeps special float values and rejects timestamps") {
+ assert(parseField(2.5d, DOUBLE) == 2.5d)
+ assert(parseField(" -0.75 ", DOUBLE) == -0.75d)
+ assert(parseField("NaN", DOUBLE).asInstanceOf[java.lang.Double].isNaN)
+ assert(parseField("Infinity", DOUBLE) == java.lang.Double.POSITIVE_INFINITY)
+ assert(parseField(false, DOUBLE) == 0.0d)
+
+ val ex = intercept[AttributeTypeException] {
+ parseField(new Timestamp(0L), DOUBLE)
+ }
+ assert(ex.getMessage.startsWith("Failed to parse type java.sql.Timestamp to Double:"))
+ }
+
+ test("parseField to BOOLEAN maps non-zero numbers to true") {
+ assert(parseField(2L, BOOLEAN) == true)
+ assert(parseField(0L, BOOLEAN) == false)
+ assert(parseField(0.5d, BOOLEAN) == true)
+ assert(parseField(0.0d, BOOLEAN) == false)
+ assert(parseField(true, BOOLEAN) == true)
+ assert(parseField(-1, BOOLEAN) == true)
+ }
+
+ test("parseField to BOOLEAN treats only the literal 1 as a true numeric string") {
+ // The fallback is `trim.toInt == 1`, so any other integer string parses to false
+ // rather than failing.
+ assert(parseField(" 1 ", BOOLEAN) == true)
+ assert(parseField("2", BOOLEAN) == false)
+ assert(parseField("-1", BOOLEAN) == false)
+ // A non-boolean, non-integer string still fails.
+ assertThrows[AttributeTypeException](parseField("yes", BOOLEAN))
+ }
+
+ test("parseField to BOOLEAN rejects timestamp input") {
+ val ex = intercept[AttributeTypeException] {
+ parseField(new Timestamp(0L), BOOLEAN)
+ }
+ assert(ex.getMessage.startsWith("Failed to parse type java.sql.Timestamp to Boolean:"))
+ }
+
+ test("parseField to TIMESTAMP rejects integers and booleans") {
+ assertThrows[AttributeTypeException](parseField(123, TIMESTAMP))
+ assertThrows[AttributeTypeException](parseField(true, TIMESTAMP))
+ // Blank strings carry no date at all.
+ assertThrows[AttributeTypeException](parseField(" ", TIMESTAMP))
+ }
+
+ test("parseField to LARGE_BINARY rejects a value that is not an s3 URI") {
+ // LargeBinary validates its URI scheme, so the failure is an IllegalArgumentException
+ // raised by the constructor rather than an AttributeTypeException.
+ val ex = intercept[IllegalArgumentException] {
+ parseField("/local/path", LARGE_BINARY)
+ }
+ assert(ex.getMessage == "LargeBinary URI must start with 's3://', got: /local/path")
+ assert(!ex.isInstanceOf[AttributeTypeException])
+ }
+
+ test("parseField to STRING and BINARY passes values through untouched") {
+ val binary = Array[Byte](9, 8, 7)
+ // BINARY is returned by reference, not copied or re-encoded.
+ assert(parseField(binary, BINARY).asInstanceOf[Array[Byte]] eq binary)
+ // STRING relies on toString, including for timestamps.
+ assert(parseField(new Timestamp(0L), STRING) == new Timestamp(0L).toString)
+ assert(parseField(Seq(1, 2), ANY) == Seq(1, 2))
+ }
+
+ test("inferSchemaFromRows returns no types for an empty iterator") {
+ assert(inferSchemaFromRows(Iterator.empty).isEmpty)
+ }
+
+ test("inferSchemaFromRows widens a column only in one direction") {
+ // Once a column has widened to STRING it never narrows back, even if later rows
+ // would parse as integers on their own.
+ val rows: Iterator[Array[Any]] = Iterator(
+ Array[Any]("1", "1"),
+ Array[Any]("not-a-number-or-date", "2"),
+ Array[Any]("3", "3")
+ )
+ val attributeTypes = inferSchemaFromRows(rows)
+ assert(attributeTypes(0) == STRING)
+ assert(attributeTypes(1) == INTEGER)
+ }
+
+ test("inferSchemaFromRows keeps INTEGER for a column of only nulls") {
+ // Null carries no evidence, so the seed type survives.
+ val rows: Iterator[Array[Any]] = Iterator(Array[Any](null), Array[Any](null))
+ assert(inferSchemaFromRows(rows).sameElements(Array(INTEGER)))
+ }
+
+ test("inferField with no prior type falls back through the type ladder") {
+ assert(inferField(null) == INTEGER)
+ assert(inferField("9223372036854775807") == LONG)
+ assert(inferField("not parseable at all") == STRING)
+ }
+
+ test("inferField never narrows a STRING column back to a parseable type") {
+ assert(inferField(STRING, "1") == STRING)
+ assert(inferField(STRING, null) == STRING)
+ }
+
+ test("compare rejects the ANY type") {
+ val ex = intercept[UnsupportedOperationException] {
+ compare(1, 2, ANY)
+ }
+ // ANY renders as the empty string (it is hidden from the JSON schema), so the message
+ // names no type.
+ assert(ex.getMessage == "Unsupported attribute type for compare: ")
+ }
+
+ test("compare orders binary arrays of differing lengths by prefix") {
+ // A prefix sorts before the longer array that extends it.
+ assert(compare(Array[Byte](1, 2), Array[Byte](1, 2, 0), BINARY) < 0)
+ assert(compare(Array.emptyByteArray, Array[Byte](0), BINARY) < 0)
+ }
+
+ test("compare places NaN above every other double") {
+ assert(compare(java.lang.Double.NaN, java.lang.Double.MAX_VALUE, DOUBLE) > 0)
+ assert(compare(java.lang.Double.NEGATIVE_INFINITY, 0.0d, DOUBLE) < 0)
+ }
+
+ test("add with two null operands fails for types that have no zero value") {
+ // The null/null branch delegates to zeroValue, which rejects unsupported types.
+ val ex = intercept[UnsupportedOperationException] {
+ add(null, null, STRING)
+ }
+ assert(ex.getMessage == "Unsupported attribute type for zero value: string")
+ }
+
+ test("add returns the non-null operand unchanged for otherwise unsupported types") {
+ // The identity branches run before the per-type dispatch, so no exception is raised.
+ assert(add("kept", null, STRING) == "kept")
+ assert(add(null, "kept", STRING) == "kept")
+ }
+
+ test("SchemaCasting leaves the schema untouched when no attribute matches") {
+ val schema = Schema(List(new Attribute("a", STRING), new Attribute("b", STRING)))
+ assert(SchemaCasting(schema, "missing", INTEGER) == schema)
+ }
+
+ test("SchemaCasting retains the original type for LARGE_BINARY targets") {
+ // LARGE_BINARY is outside the supported cast list, so the attribute keeps its type.
+ val schema = Schema(List(new Attribute("a", STRING)))
+ assert(SchemaCasting(schema, "a", LARGE_BINARY) == schema)
+ }
+
+ test("SchemaCasting supports every castable target type") {
+ val schema = Schema(List(new Attribute("a", STRING)))
+ Seq(STRING, INTEGER, DOUBLE, LONG, BOOLEAN, TIMESTAMP, BINARY).foreach { target =>
+ assert(SchemaCasting(schema, "a", target).getAttribute("a").getType == target)
+ }
+ }
+
+ test("tupleCasting preserves nulls and leaves unlisted columns alone") {
+ val schema = Schema(List(new Attribute("a", STRING), new Attribute("b", STRING)))
+ val tuple = Tuple.builder(schema).addSequentially(Array[Any](null, "42")).build()
+ val result = tupleCasting(tuple, Map("a" -> INTEGER))
+ assert(result.getFields.sameElements(Array[Any](null, "42")))
+ }
+
+ test("tupleCasting propagates a parse failure for an uncastable value") {
+ val schema = Schema(List(new Attribute("a", STRING)))
+ val tuple = Tuple.builder(schema).addSequentially(Array[Any]("abc")).build()
+ assertThrows[AttributeTypeException](tupleCasting(tuple, Map("a" -> INTEGER)))
+ }
+
+ test("parseFields returns an empty array for empty input") {
+ assert(parseFields(Array.empty[Any], Array.empty[AttributeType]).isEmpty)
+ }
}
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/util/StackTraceUtilsSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/StackTraceUtilsSpec.scala
new file mode 100644
index 0000000000..afb23300d3
--- /dev/null
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/StackTraceUtilsSpec.scala
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.util
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class StackTraceUtilsSpec extends AnyFunSuite {
+
+ private val TopLevelHeader = "Stack trace for developers: \n\n"
+ private val CauseHeader = "\n\nCaused by:\n"
+
+ /** Builds a throwable with no frames so the rendering can be asserted exactly. */
+ private def frameless(err: Throwable): Throwable = {
+ err.setStackTrace(Array.empty)
+ err
+ }
+
+ private def occurrencesOf(haystack: String, needle: String): Int =
+ haystack.sliding(needle.length).count(_ == needle)
+
+ test("getStackTraceWithAllCauses renders a causeless throwable exactly") {
+ val err = frameless(new RuntimeException("boom"))
+
+ // header + toString + "\n" + (no frames) is the whole output.
+ assert(
+ StackTraceUtils.getStackTraceWithAllCauses(err) ==
+ s"${TopLevelHeader}java.lang.RuntimeException: boom\n"
+ )
+ }
+
+ test("getStackTraceWithAllCauses includes every real stack frame, newline-separated") {
+ val err = new RuntimeException("with frames")
+ val rendered = StackTraceUtils.getStackTraceWithAllCauses(err)
+
+ assert(rendered.startsWith(TopLevelHeader))
+ assert(rendered.contains("java.lang.RuntimeException: with frames"))
+ // Frames are rendered verbatim, in order, after the throwable's own toString.
+ val frameBlock = rendered.stripPrefix(s"${TopLevelHeader}${err.toString}\n")
+ assert(frameBlock == err.getStackTrace.mkString("\n"))
+ assert(err.getStackTrace.length > 1)
+ assert(frameBlock.contains(this.getClass.getName))
+ }
+
+ test("getStackTraceWithAllCauses omits the Caused-by section when there is no cause") {
+ val rendered = StackTraceUtils.getStackTraceWithAllCauses(new RuntimeException("lonely"))
+ assert(!rendered.contains("Caused by:"))
+ }
+
+ test("getStackTraceWithAllCauses with topLevel=false swaps in the Caused-by header") {
+ val err = frameless(new IllegalStateException("nested"))
+
+ assert(
+ StackTraceUtils.getStackTraceWithAllCauses(err, topLevel = false) ==
+ s"${CauseHeader}java.lang.IllegalStateException: nested\n"
+ )
+ assert(
+ !StackTraceUtils
+ .getStackTraceWithAllCauses(err, topLevel = false)
+ .contains("Stack trace for developers")
+ )
+ }
+
+ test("getStackTraceWithAllCauses appends a single cause below the top-level trace") {
+ val cause = frameless(new IllegalArgumentException("inner"))
+ val err = frameless(new RuntimeException("outer", cause))
+
+ assert(
+ StackTraceUtils.getStackTraceWithAllCauses(err) ==
+ s"${TopLevelHeader}java.lang.RuntimeException: outer\n" +
+ s"${CauseHeader}java.lang.IllegalArgumentException: inner\n"
+ )
+ }
+
+ test("getStackTraceWithAllCauses walks the whole chain, outermost first") {
+ val root = frameless(new java.io.IOException("root"))
+ val middle = frameless(new IllegalStateException("middle", root))
+ val top = frameless(new RuntimeException("top", middle))
+
+ val rendered = StackTraceUtils.getStackTraceWithAllCauses(top)
+
+ // One developer header for the whole chain, one Caused-by per cause.
+ assert(occurrencesOf(rendered, "Stack trace for developers") == 1)
+ assert(occurrencesOf(rendered, "Caused by:") == 2)
+ // Ordering is outermost -> innermost.
+ assert(rendered.indexOf("top") < rendered.indexOf("middle"))
+ assert(rendered.indexOf("middle") < rendered.indexOf("root"))
+ assert(rendered.endsWith("java.io.IOException: root\n"))
+ }
+
+ test("getStackTraceWithAllCauses renders each cause's own frames, not just the top's") {
+ val cause = new IllegalStateException("inner")
+ val err = new RuntimeException("outer", cause)
+
+ val rendered = StackTraceUtils.getStackTraceWithAllCauses(err)
+
+ assert(rendered.contains(err.getStackTrace.head.toString))
+ assert(rendered.contains(cause.getStackTrace.head.toString))
+ // The cause's frames belong to the section that follows the Caused-by marker.
+ val causeSection = rendered.substring(rendered.indexOf(CauseHeader))
+ assert(causeSection.contains(cause.getStackTrace.head.toString))
+ }
+
+ test("getStackTraceWithAllCauses handles a throwable without a message") {
+ val err = frameless(new RuntimeException())
+
+ // Throwable.toString drops the ": <msg>" suffix when the message is null.
+ assert(
+ StackTraceUtils.getStackTraceWithAllCauses(err) ==
+ s"${TopLevelHeader}java.lang.RuntimeException\n"
+ )
+ }
+
+ test("getStackTraceWithAllCauses renders a deep cause chain without truncating it") {
+ // The recursion is unbounded by design; a long chain must be rendered in full.
+ val root = frameless(new RuntimeException("root"))
+ val chain = (1 to 20).foldLeft[Throwable](root) { (cause, i) =>
+ frameless(new RuntimeException(s"level-$i", cause))
+ }
+
+ val rendered = StackTraceUtils.getStackTraceWithAllCauses(chain)
+
+ assert(occurrencesOf(rendered, "Caused by:") == 20)
+ assert(rendered.startsWith(TopLevelHeader))
+ assert(rendered.endsWith("java.lang.RuntimeException: root\n"))
+ }
+}
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/service/util/LargeBinaryManagerSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/service/util/LargeBinaryManagerSpec.scala
index 0f2cce8b30..eeb1bce7cd 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/service/util/LargeBinaryManagerSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/service/util/LargeBinaryManagerSpec.scala
@@ -523,4 +523,80 @@ class LargeBinaryManagerSpec extends AnyFunSuite with S3StorageTestBase with Bef
setExecutionContext(TestExecutionId)
}
}
+
+ // ========================================
+ // Base-URI Tests
+ // ========================================
+
+ test("baseUriForExecution scopes binaries to the configured bucket and the execution") {
+ assert(
+ LargeBinaryManager.baseUriForExecution(4242L) ==
+ s"s3://${LargeBinaryManager.DEFAULT_BUCKET}/objects/4242/"
+ )
+ // The trailing slash is what keeps execution 4242 and 42421 in separate prefixes.
+ assert(LargeBinaryManager.baseUriForExecution(4242L).endsWith("/"))
+ assert(LargeBinaryManager.DEFAULT_BUCKET == "texera-large-binaries")
+ }
+
+ test("re-seeding the base URI routes later binaries to the new execution's prefix") {
+ setExecutionContext(3001L)
+ val first = createLargeBinary("written under 3001")
+ setExecutionContext(3002L)
+ val second = createLargeBinary("written under 3002")
+
+ try {
+ assert(first.getObjectKey.startsWith("objects/3001/"))
+ assert(second.getObjectKey.startsWith("objects/3002/"))
+ // Both objects really landed under their own execution prefix in S3.
+ assert(S3StorageClient.directoryExists(LargeBinaryManager.DEFAULT_BUCKET, "objects/3001"))
+ assert(S3StorageClient.directoryExists(LargeBinaryManager.DEFAULT_BUCKET, "objects/3002"))
+ } finally {
+ LargeBinaryManager.deleteByExecution(3001L)
+ LargeBinaryManager.deleteByExecution(3002L)
+ setExecutionContext(TestExecutionId)
+ }
+ }
+
+ test("an empty base URI clears the thread's context so create fails loudly") {
+ // An unconfigured bucket yields an empty base URI; create() must not silently produce
+ // a bucket-less key.
+ try {
+ LargeBinaryManager.setCurrentBaseUri("")
+ val thrown = intercept[IllegalStateException](LargeBinaryManager.create())
+ assert(thrown.getMessage.contains("requires a base URI"))
+ } finally {
+ setExecutionContext(TestExecutionId)
+ }
+ }
+
+ test("a null base URI clears the thread's context so create fails loudly") {
+ try {
+ LargeBinaryManager.setCurrentBaseUri(null)
+ assertThrows[IllegalStateException](LargeBinaryManager.create())
+ // new LargeBinary() delegates to create(), so it fails the same way.
+ assertThrows[IllegalStateException](new LargeBinary())
+ } finally {
+ setExecutionContext(TestExecutionId)
+ }
+ }
+
+ test("the base URI is thread-local and does not leak into other threads") {
+ // create() runs on the DP thread; a sibling thread must not inherit this thread's context.
+ setExecutionContext(TestExecutionId)
+ @volatile var caught: Option[Throwable] = None
+ val other = new Thread(() => {
+ try LargeBinaryManager.create()
+ catch { case e: Throwable => caught = Some(e) }
+ })
+ other.start()
+ other.join()
+
+ assert(caught.exists(_.isInstanceOf[IllegalStateException]))
+ // This thread's context is unaffected.
+ assert(
+ LargeBinaryManager
+ .create()
+ .startsWith(LargeBinaryManager.baseUriForExecution(TestExecutionId))
+ )
+ }
}
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/service/util/S3StorageClientSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/service/util/S3StorageClientSpec.scala
index c98f3c589c..bcd50dc23d 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/service/util/S3StorageClientSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/service/util/S3StorageClientSpec.scala
@@ -19,14 +19,19 @@
package org.apache.texera.service.util
+import org.apache.texera.common.config.StorageConfig
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
import org.scalatest.funsuite.AnyFunSuite
-import software.amazon.awssdk.services.s3.model.S3Error
+import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.s3.model._
+import software.amazon.awssdk.services.s3.{S3Client, S3Configuration}
-import java.io.ByteArrayInputStream
+import java.io.{ByteArrayInputStream, IOException, InputStream}
import java.util.concurrent.Executors
import scala.concurrent.duration._
import scala.concurrent.{Await, ExecutionContext, Future}
+import scala.jdk.CollectionConverters._
import scala.util.Random
class S3StorageClientSpec
@@ -45,13 +50,64 @@ class S3StorageClientSpec
override def afterAll(): Unit = {
// Best-effort cleanup of the prefixes these tests use (deleteDirectory rejects an empty prefix).
try {
- Seq("test", "delete-dir").foreach(S3StorageClient.deleteDirectory(testBucketName, _))
+ Seq("test", "delete-dir", "multipart")
+ .foreach(S3StorageClient.deleteDirectory(testBucketName, _))
} catch {
case _: Exception => // Ignore cleanup errors
}
+ try rawClient.close()
+ catch { case _: Exception => }
super.afterAll()
}
+ /**
+ * A second S3 client pointed at the same MinIO container. S3StorageClient exposes only
+ * `uploadPartWithRequest` from the multipart API, so the surrounding create/complete/list
+ * calls are issued directly instead of being mocked away.
+ */
+ private lazy val rawClient: S3Client = {
+ val credentials = AwsBasicCredentials.create(StorageConfig.s3Username, StorageConfig.s3Password)
+ S3Client
+ .builder()
+ .credentialsProvider(StaticCredentialsProvider.create(credentials))
+ .region(Region.of(StorageConfig.s3Region))
+ .endpointOverride(java.net.URI.create(StorageConfig.s3Endpoint))
+ .serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(true).build())
+ .build()
+ }
+
+ private def startMultipartUpload(objectKey: String): String =
+ rawClient
+ .createMultipartUpload(
+ CreateMultipartUploadRequest.builder().bucket(testBucketName).key(objectKey).build()
+ )
+ .uploadId()
+
+ private def completeMultipartUpload(
+ objectKey: String,
+ uploadId: String,
+ parts: Seq[CompletedPart]
+ ): Unit =
+ rawClient.completeMultipartUpload(
+ CompleteMultipartUploadRequest
+ .builder()
+ .bucket(testBucketName)
+ .key(objectKey)
+ .uploadId(uploadId)
+ .multipartUpload(CompletedMultipartUpload.builder().parts(parts.asJava).build())
+ .build()
+ )
+
+ /** Upload ids still pending (i.e. neither completed nor aborted) for the given key. */
+ private def pendingUploadIds(objectKey: String): Seq[String] =
+ rawClient
+ .listMultipartUploads(ListMultipartUploadsRequest.builder().bucket(testBucketName).build())
+ .uploads()
+ .asScala
+ .filter(_.key() == objectKey)
+ .map(_.uploadId())
+ .toSeq
+
// Helper methods
private def createInputStream(data: String): ByteArrayInputStream = {
new ByteArrayInputStream(data.getBytes)
@@ -438,4 +494,226 @@ class S3StorageClientSpec
assert(!thrown.getMessage.contains(f"delete-dir/locked-$cap%02d.txt")) // capped key is not
assert(thrown.getMessage.contains(s"and ${errorCount - cap} more"))
}
+
+ test("deleteDirectory should reject an empty prefix instead of wiping the bucket") {
+ val guard = "delete-dir/guarded-object.txt"
+ S3StorageClient.uploadObject(testBucketName, guard, createInputStream("keep me"))
+
+ val thrown = intercept[IllegalArgumentException] {
+ S3StorageClient.deleteDirectory(testBucketName, "")
+ }
+ assert(thrown.getMessage.contains("directoryPrefix must not be empty"))
+
+ // The bucket is untouched.
+ val survivor = S3StorageClient.downloadObject(testBucketName, guard)
+ assert(new String(readInputStream(survivor)) == "keep me")
+ survivor.close()
+ S3StorageClient.deleteObject(testBucketName, guard)
+ }
+
+ // ========================================
+ // directoryExists Tests
+ // ========================================
+
+ test("directoryExists should treat a prefix with and without a trailing slash the same") {
+ val prefix = "test/exists-dir"
+ S3StorageClient.uploadObject(
+ testBucketName,
+ s"$prefix/object.txt",
+ createInputStream("data")
+ )
+
+ assert(S3StorageClient.directoryExists(testBucketName, prefix))
+ assert(S3StorageClient.directoryExists(testBucketName, s"$prefix/"))
+
+ S3StorageClient.deleteObject(testBucketName, s"$prefix/object.txt")
+ }
+
+ test("directoryExists should be false for a prefix that only matches an object name") {
+ // "test/exists-file" names an object, not a directory, so no key starts with
+ // "test/exists-file/".
+ val objectKey = "test/exists-file"
+ S3StorageClient.uploadObject(testBucketName, objectKey, createInputStream("data"))
+
+ assert(!S3StorageClient.directoryExists(testBucketName, objectKey))
+
+ S3StorageClient.deleteObject(testBucketName, objectKey)
+ }
+
+ test("directoryExists should be false for an unused prefix") {
+ assert(!S3StorageClient.directoryExists(testBucketName, "test/never-written"))
+ }
+
+ // ========================================
+ // createBucketIfNotExist Tests
+ // ========================================
+
+ test("createBucketIfNotExist should be idempotent for an existing bucket") {
+ // beforeAll already created it; a second call takes the headBucket-succeeds path.
+ S3StorageClient.createBucketIfNotExist(testBucketName)
+ S3StorageClient.createBucketIfNotExist(testBucketName)
+
+ // The bucket is still usable afterwards.
+ val objectKey = "test/after-recreate.txt"
+ S3StorageClient.uploadObject(testBucketName, objectKey, createInputStream("still here"))
+ val stream = S3StorageClient.downloadObject(testBucketName, objectKey)
+ assert(new String(readInputStream(stream)) == "still here")
+ stream.close()
+ S3StorageClient.deleteObject(testBucketName, objectKey)
+ }
+
+ // ========================================
+ // uploadPartWithRequest Tests
+ // ========================================
+
+ test("uploadPartWithRequest should stream a part when the content length is known") {
+ val objectKey = "multipart/known-length.bin"
+ val partData = "part payload with a known length".getBytes
+ val uploadId = startMultipartUpload(objectKey)
+
+ val response = S3StorageClient.uploadPartWithRequest(
+ testBucketName,
+ objectKey,
+ uploadId,
+ partNumber = 1,
+ createInputStream(partData),
+ contentLength = Some(partData.length.toLong)
+ )
+
+ assert(response.eTag() != null && response.eTag().nonEmpty)
+ completeMultipartUpload(
+ objectKey,
+ uploadId,
+ Seq(CompletedPart.builder().partNumber(1).eTag(response.eTag()).build())
+ )
+
+ val stream = S3StorageClient.downloadObject(testBucketName, objectKey)
+ assert(readInputStream(stream).sameElements(partData))
+ stream.close()
+ S3StorageClient.deleteObject(testBucketName, objectKey)
+ }
+
+ test("uploadPartWithRequest should buffer the whole stream when no content length is given") {
+ val objectKey = "multipart/unknown-length.bin"
+ val partData = "part payload with no declared length".getBytes
+ val uploadId = startMultipartUpload(objectKey)
+
+ val response = S3StorageClient.uploadPartWithRequest(
+ testBucketName,
+ objectKey,
+ uploadId,
+ partNumber = 1,
+ createInputStream(partData),
+ contentLength = None
+ )
+
+ completeMultipartUpload(
+ objectKey,
+ uploadId,
+ Seq(CompletedPart.builder().partNumber(1).eTag(response.eTag()).build())
+ )
+
+ val stream = S3StorageClient.downloadObject(testBucketName, objectKey)
+ assert(readInputStream(stream).sameElements(partData))
+ stream.close()
+ S3StorageClient.deleteObject(testBucketName, objectKey)
+ }
+
+ test("uploadPartWithRequest should assemble parts in part-number order") {
+ // Two parts: every part but the last must be at least 5 MiB, so the head is sized exactly
+ // at the multipart minimum and the tail is short.
+ val objectKey = "multipart/ordered-parts.bin"
+ val head = Array.fill[Byte](S3StorageClient.MINIMUM_NUM_OF_MULTIPART_S3_PART.toInt)(1.toByte)
+ val tail = Array.fill[Byte](1024)(2.toByte)
+ val uploadId = startMultipartUpload(objectKey)
+
+ val firstETag = S3StorageClient
+ .uploadPartWithRequest(
+ testBucketName,
+ objectKey,
+ uploadId,
+ partNumber = 1,
+ createInputStream(head),
+ contentLength = Some(head.length.toLong)
+ )
+ .eTag()
+ val secondETag = S3StorageClient
+ .uploadPartWithRequest(
+ testBucketName,
+ objectKey,
+ uploadId,
+ partNumber = 2,
+ createInputStream(tail),
+ contentLength = None
+ )
+ .eTag()
+
+ completeMultipartUpload(
+ objectKey,
+ uploadId,
+ Seq(
+ CompletedPart.builder().partNumber(1).eTag(firstETag).build(),
+ CompletedPart.builder().partNumber(2).eTag(secondETag).build()
+ )
+ )
+
+ val stream = S3StorageClient.downloadObject(testBucketName, objectKey)
+ val downloaded = readInputStream(stream)
+ stream.close()
+
+ assert(downloaded.length == head.length + tail.length)
+ assert(downloaded.take(head.length).sameElements(head))
+ assert(downloaded.drop(head.length).sameElements(tail))
+
+ S3StorageClient.deleteObject(testBucketName, objectKey)
+ }
+
+ test("uploadPartWithRequest should fail for an unknown upload id") {
+ val objectKey = "multipart/no-such-upload.bin"
+ val thrown = intercept[S3Exception] {
+ S3StorageClient.uploadPartWithRequest(
+ testBucketName,
+ objectKey,
+ uploadId = "not-a-real-upload-id",
+ partNumber = 1,
+ createInputStream("payload"),
+ contentLength = None
+ )
+ }
+ assert(thrown.statusCode() == 404)
+ }
+
+ // ========================================
+ // uploadObject failure-path Tests
+ // ========================================
+
+ test("uploadObject should abort the multipart upload when the source stream fails") {
+ // Fails only after the first full part has been read, so a multipart upload is already
+ // in flight when the error surfaces.
+ val objectKey = "multipart/failing-source.bin"
+ val firstPart =
+ Array.fill[Byte](S3StorageClient.MINIMUM_NUM_OF_MULTIPART_S3_PART.toInt)(7.toByte)
+
+ val failingStream = new InputStream {
+ private val delegate = new ByteArrayInputStream(firstPart)
+ override def read(): Int = throw new IOException("stream broke")
+ override def read(b: Array[Byte], off: Int, len: Int): Int = {
+ val n = delegate.read(b, off, len)
+ if (n <= 0) throw new IOException("stream broke") else n
+ }
+ }
+
+ val thrown = intercept[IOException] {
+ S3StorageClient.uploadObject(testBucketName, objectKey, failingStream)
+ }
+ assert(thrown.getMessage == "stream broke")
+
+ // The in-flight upload was aborted, so no orphaned upload is left behind...
+ assert(pendingUploadIds(objectKey).isEmpty)
+ // ...and no object was published under the key.
+ val missing = intercept[S3Exception] {
+ S3StorageClient.downloadObject(testBucketName, objectKey)
+ }
+ assert(missing.statusCode() == 404)
+ }
}