Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,9 @@ class URLFetcherOpExec(descString: String) extends SourceOperatorExecutor {
override def produceTuple(): Iterator[TupleLike] = {

val urlObj = new URL(desc.url)
val input = getInputStreamFromURL(urlObj)
val contentInputStream = input match {
case Some(value) => value
case None => IOUtils.toInputStream(s"Fetch failed for URL: $desc.url", "UTF-8")
}
val contentInputStream = getInputStreamFromURL(urlObj).getOrElse(
IOUtils.toInputStream(s"Fetch failed for URL: ${desc.url}", "UTF-8")
)
Iterator(if (desc.decodingMethod == DecodingMethod.UTF_8) {
TupleLike(IOUtils.toString(contentInputStream, "UTF-8"))
} else {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -44,4 +44,27 @@ class URLFetcherOpExecSpec extends AnyFlatSpec with BeforeAndAfter {
assert(!iterator.hasNext)
}

// On a failed fetch the fallback message must interpolate `desc.url` itself,
// not the descriptor's reflectionToString dump. A file:// URL to a nonexistent
// path makes getInputStreamFromURL return None deterministically and offline,
// so the failure branch is exercised without depending on external connectivity.
it should "report only the URL, not the operator descriptor, when the fetch fails" in {
val missingUrl =
java.nio.file.Files
.createTempDirectory("texera-urlfetcher-regression-")
.resolve("missing")
.toUri
.toString
opDesc.url = missingUrl
opDesc.decodingMethod = DecodingMethod.UTF_8
val fetcherOpExec = new URLFetcherOpExec(objectMapper.writeValueAsString(opDesc))
val content = fetcherOpExec.produceTuple().next().getFields.toList.head.asInstanceOf[String]

assert(content == s"Fetch failed for URL: ${opDesc.url}")
// Guard against the pre-fix `$desc.url` behavior, which leaked the whole
// descriptor dump (class name + internal fields) into the message.
assert(!content.contains("URLFetcherOpDesc"))
assert(!content.contains("operatorId"))
}

}
Loading