Bora Elkin (BoraElkin) · Nov 6, 2026 · Nov 6, 2026 · Nov 6, 2026 · Nov 7, 2026 · Nov 7, 2026
@@ -3,9 +3,9 @@ codecov:
     require_ci_to_pass: no
     # Wait for expected number of coverage uploads before sending notifications
     # This prevents premature comments with incomplete data.
-    # 55 = UnitTests 41 + PythonTests 7 + RTests 6 + WebsiteSamplesTests 1
+    # 56 = UnitTests 42 + PythonTests 7 + RTests 6 + WebsiteSamplesTests 1
     # (every leg of those jobs runs templates/codecov.yml on succeededOrFailed).
-    after_n_builds: 55
+    after_n_builds: 56
 
 comment:
   layout: "reach,diff,flags,files"
@@ -14,7 +14,7 @@ comment:
   require_base: false
   require_head: true
   # Wait for all expected uploads before commenting (see note above)
-  after_n_builds: 55
+  after_n_builds: 56
 
 coverage:
   precision: 2

@@ -223,6 +223,12 @@ trait HasCustomHeaders extends HasServiceParams {
     setScalarParam(customHeaders, ServiceAuthHeaders.sanitizeHeaderMap(v))
   }
 
+  def getCustomHeaders: Map[String, String] = getScalarParam(customHeaders)
+
+  def setCustomHeadersCol(value: String): this.type = setVectorParam(customHeaders, value)
+
+  def getCustomHeadersCol: String = getVectorParam(customHeaders)
+
   // For Pyspark compatibility accept Java HashMap as input to parameter
   // py4J only natively supports conversions from Python Dict to Java HashMap. A null HashMap is
   // handled here (never dereferenced) and null keys/values are dropped by the Scala overload above.
@@ -351,6 +357,11 @@ private[ml] object ServiceAuthHeaders {
   private def sanitizeHeaderMap(headers: Option[Map[String, String]]): Map[String, String] =
     sanitizeHeaderMap(Option(headers).flatten.orNull)
 
+  private def addContentType(headers: mutable.Map[String, String], value: String): Unit = {
+    headers.keys.filter(_.equalsIgnoreCase("Content-Type")).toSeq.foreach(headers.remove)
+    headers += ("Content-Type" -> value)
+  }
+
   // Deterministically select a non-blank credential embedded in customHeaders for one canonical
   // auth header name, canonicalizing its casing. Blank values are ignored and mixed-case duplicates
   // resolve by sorted header name, so an empty entry can never suppress a valid credential.
@@ -414,7 +425,7 @@ private[ml] object ServiceAuthHeaders {
         headers += (headerName -> headerValue)
       }
     }
-    contentType.filterNot(StringUtils.isEmpty).foreach(value => headers += ("Content-Type" -> value))
+    contentType.filterNot(StringUtils.isEmpty).foreach(value => addContentType(headers, value))
 
     Resolution(
       new scala.collection.immutable.TreeMap[String, String]() ++ headers,

@@ -0,0 +1,56 @@
+// Copyright (C) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See LICENSE in project root for information.
+
+package com.microsoft.azure.synapse.ml.services.contentunderstanding
+
+import java.util.concurrent.TimeUnit
+import scala.annotation.tailrec
+import scala.concurrent.blocking
+
+private[contentunderstanding] object ContentUnderstandingBudget {
+  val CancellationPollMs: Long = 50L
+}
+
+/** Thread-confined operation budget; concurrent rows must use separate instances. */
+private[contentunderstanding] class ContentUnderstandingBudget(
+    durationMs: Int, nanoTime: () => Long = () => System.nanoTime()) {
+  private var started = nanoTime()
+
+  def mark(): Long = nanoTime()
+
+  // Exclude only the interval since this mark; previously charged time remains charged.
+  def refundSince(mark: Long): Unit = {
+    started += nanoTime() - mark
+  }
+
+  def remainingMs: Long = math.max(0L,
+    durationMs.toLong - TimeUnit.NANOSECONDS.toMillis(nanoTime() - started))
+
+  def pause(delay: Long): Unit = pause(delay, () => ())
+
+  // Tick failures propagate to the caller, allowing a retry's shared-stop signal to abort this wait.
+  def pause(delay: Long, onTick: () => Unit): Unit = blocking {
+    ContentUnderstandingProtocol.checkInterrupted()
+    onTick()
+    if (delay > 0) {
+      try {
+        val start = System.nanoTime()
+        @tailrec
+        def sleep(): Unit = {
+          ContentUnderstandingProtocol.checkInterrupted()
+          onTick()
+          val remaining = delay - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)
+          if (remaining > 0) {
+            Thread.sleep(math.min(remaining, ContentUnderstandingBudget.CancellationPollMs))
+            sleep()
+          }
+        }
+        sleep()
+      } catch {
+        case error: InterruptedException =>
+          Thread.currentThread().interrupt()
+          throw error
+      }
+    }
+  }
+}
@@ -0,0 +1,174 @@
+// Copyright (C) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See LICENSE in project root for information.
+
+package com.microsoft.azure.synapse.ml.services.contentunderstanding
+
+import org.apache.http.conn.DnsResolver
+
+import java.io.InterruptedIOException
+import java.net.{InetAddress, UnknownHostException}
+import java.util.ArrayDeque
+import java.util.concurrent.{ArrayBlockingQueue, Callable, ExecutionException, Future, RejectedExecutionException,
+  Semaphore, ThreadFactory, ThreadPoolExecutor, TimeUnit, TimeoutException}
+import scala.annotation.tailrec
+
+private[contentunderstanding] class OperationDeadlineExceeded
+  extends UnknownHostException(ContentUnderstandingProtocol.timeoutError)
+
+private[contentunderstanding] class DnsResolutionFailed(cause: Throwable)
+  extends UnknownHostException(ContentUnderstandingProtocol.diagnostic("DnsResolutionFailed",
+    "DNS resolution failed before the HTTP request was sent.")) {
+  initCause(cause)
+}
+
+private[contentunderstanding] class DnsCapacityExceeded(val deadlineExpired: Boolean = false)
+  extends UnknownHostException(ContentUnderstandingProtocol.diagnostic("DnsCapacityExceeded",
+    "DNS worker capacity was unavailable within the operation budget; no HTTP request was sent."))
+
+/** Bounds blocking JVM DNS resolution by the current Content Understanding operation budget. */
+private[contentunderstanding] class BudgetedDnsResolver(remainingMs: () => Long,
+                                                        delegate: DnsResolver) extends DnsResolver {
+  override def resolve(host: String): Array[InetAddress] = {
+    val remaining = remainingMs()
+    if (remaining == 0) {
+      throw new OperationDeadlineExceeded
+    }
+    try {
+      BudgetedDnsResolver.acquire(remaining)
+    } catch {
+      case error: InterruptedException => throw interrupted(error)
+    }
+    var submitted = false
+    val resolution = try {
+      ContentUnderstandingProtocol.checkInterrupted()
+      if (remainingMs() == 0) throw new OperationDeadlineExceeded
+      val future = BudgetedDnsResolver.Executor.submit(new Callable[Array[InetAddress]] {
+        override def call(): Array[InetAddress] =
+          try {
+            delegate.resolve(host)
+          } finally {
+            BudgetedDnsResolver.release()
+          }
+      })
+      submitted = true
+      future
+    } catch {
+      case error: InterruptedException => throw interrupted(error)
+      case _: RejectedExecutionException =>
+        throw new DnsCapacityExceeded()
+    } finally {
+      if (!submitted) BudgetedDnsResolver.release()
+    }
+    awaitResolution(resolution)
+  }
+
+  private def awaitResolution(resolution: Future[Array[InetAddress]]): Array[InetAddress] = {
+    val timeout = remainingMs()
+    val started = System.nanoTime()
+    @tailrec
+    def poll(): Array[InetAddress] = {
+      ContentUnderstandingProtocol.checkInterrupted()
+      val wait = BudgetedDnsResolver.waitRemaining(started, timeout)
+      if (wait == 0) throw new OperationDeadlineExceeded
+      val addresses = try {
+        Some(resolution.get(math.min(wait, ContentUnderstandingBudget.CancellationPollMs), TimeUnit.MILLISECONDS))
+      } catch {
+        case _: TimeoutException => None
+      }
+      addresses match {
+        case Some(value) => value
+        case None => poll()
+      }
+    }
+    try {
+      val addresses = poll()
+      if (remainingMs() == 0) {
+        throw new OperationDeadlineExceeded
+      }
+      addresses
+    } catch {
+      case error: ExecutionException =>
+        throw new DnsResolutionFailed(error.getCause)
+      case error: InterruptedException =>
+        throw interrupted(error)
+    }
+  }
+
+  private def interrupted(error: InterruptedException): InterruptedIOException = {
+    Thread.currentThread().interrupt()
+    val failure = new InterruptedIOException("DNS resolution was interrupted.")
+    failure.initCause(error)
+    failure
+  }
+}
+
+private[contentunderstanding] object BudgetedDnsResolver {
+  def expiredBeforeDispatch(error: UnknownHostException): Boolean = error match {
+    case _: OperationDeadlineExceeded => true
+    case capacity: DnsCapacityExceeded => capacity.deadlineExpired
+    case _ => false
+  }
+
+  def diagnostic(error: UnknownHostException, timedOut: Boolean): String =
+    Option(error.getMessage).filter(_ =>
+      error.isInstanceOf[DnsCapacityExceeded] || error.isInstanceOf[DnsResolutionFailed] || !timedOut)
+      .getOrElse(ContentUnderstandingProtocol.timeoutError)
+
+  private[contentunderstanding] val MaxConcurrentLookups = 32
+  private val IdleThreadKeepAliveSeconds = 60L
+
+  private val Capacity = new Semaphore(MaxConcurrentLookups, true)
+  private val Waiters = new ArrayDeque[Object]()
+
+  private def waitRemaining(started: Long, timeout: Long): Long =
+    math.max(0L, timeout - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started))
+
+  private def acquire(timeout: Long): Unit = {
+    val started = System.nanoTime()
+    val waiter = new Object()
+    Waiters.synchronized { Waiters.addLast(waiter) }
+    @tailrec
+    def poll(): Unit = {
+      ContentUnderstandingProtocol.checkInterrupted()
+      val acquired = Waiters.synchronized {
+        val remaining = waitRemaining(started, timeout)
+        if (remaining == 0) throw new DnsCapacityExceeded(deadlineExpired = true)
+        if ((Waiters.peekFirst() eq waiter) && Capacity.tryAcquire(0L, TimeUnit.MILLISECONDS)) {
+          true
+        } else {
+          Waiters.wait(math.min(remaining, ContentUnderstandingBudget.CancellationPollMs))
+          false
+        }
+      }
+      if (!acquired) poll()
+    }
+    // Keep the queue position across cancellation checks, removing it on every exit path.
+    try {
+      poll()
+    } finally {
+      Waiters.synchronized {
+        Waiters.remove(waiter)
+        Waiters.notifyAll()
+      }
+    }
+  }
+
+  private def release(): Unit = Waiters.synchronized {
+    Capacity.release()
+    Waiters.notifyAll()
+  }
+
+  private lazy val Executor = {
+    val executor = new ThreadPoolExecutor(MaxConcurrentLookups, MaxConcurrentLookups, IdleThreadKeepAliveSeconds,
+    TimeUnit.SECONDS,
+    new ArrayBlockingQueue[Runnable](MaxConcurrentLookups), new ThreadFactory {
+      override def newThread(task: Runnable): Thread = {
+        val thread = new Thread(task, "synapseml-cu-dns")
+        thread.setDaemon(true)
+        thread
+      }
+    })
+    executor.allowCoreThreadTimeOut(true)
+    executor
+  }
+}
@@ -0,0 +1,30 @@
+// Copyright (C) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See LICENSE in project root for information.
+
+package com.microsoft.azure.synapse.ml.services.contentunderstanding
+
+import spray.json.JsString
+
+/** The response retains the service error and operation handle, including provisioning timeouts. */
+class ContentUnderstandingException(val response: ContentUnderstandingResponse)
+  extends IllegalStateException(ContentUnderstandingException.message(response))
+
+private[contentunderstanding] object ContentUnderstandingException {
+  private val MaxDiagnosticCharacters = 4096
+  private val DiagnosticCodes = Set(
+    "AnalyzerNotReady", "AnalyzerRetrievalFailed", "DnsCapacityExceeded", "DnsResolutionFailed",
+    "HttpError", "InvalidInput", "InvalidOperationLocation", "InvalidResponse", "MissingOperationLocation",
+    "OperationFailed", "OperationLookupFailed", "OperationTimedOut", "PollAttemptsExhausted", "RequestNotSent",
+    "ResponseBodyReadFailed", "ResponseTooLarge", "SharedConfigurationFailure", "TransportError",
+    "UnacceptedOperationLocation")
+
+  private def message(response: ContentUnderstandingResponse): String = {
+    val code = response.error.filter(_.length <= MaxDiagnosticCharacters)
+      .flatMap(ContentUnderstandingResponseParsing.asObject).flatMap(_.fields.get("code"))
+      .collect { case JsString(value) if DiagnosticCodes(value) => value }
+    val diagnostic = code.map(value => s" [$value]").getOrElse("")
+    s"Content Understanding could not complete the call (status ${response.status}, HTTP ${response.httpStatus})" +
+      s"$diagnostic. Inspect the exception response for the diagnostic and any retained operation handle. " +
+      "Service details are not included in this message."
+  }
+}
@@ -0,0 +1,26 @@
+// Copyright (C) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See LICENSE in project root for information.
+
+package com.microsoft.azure.synapse.ml.services.contentunderstanding
+
+private[contentunderstanding] object ContentUnderstandingHeaderValidation {
+  private val ConnectorManagedHeaders = Set("content-length", "content-type", "host", "transfer-encoding")
+  private val HeaderToken = "[!#$%&'*+.^_`|~0-9A-Za-z-]+"
+
+  private def safeHeaderValue(value: String): Boolean =
+    !value.exists(c => c > 255 || (Character.isISOControl(c) && c != '\t'))
+
+  /** Rejects header injection and connector-managed headers without echoing the value. */
+  def validateSetting(name: String, value: Any): Unit = value match {
+    case text: String =>
+      require(safeHeaderValue(text), s"$name cannot contain control characters or characters outside Latin-1.")
+    case headers: Map[_, _] => headers.foreach { case (key, item) =>
+      val header = key.toString
+      require(header.matches(HeaderToken) && !ConnectorManagedHeaders(header.toLowerCase(java.util.Locale.ROOT)),
+        s"$name contains an invalid or connector-managed header name.")
+      require(safeHeaderValue(item.toString),
+        s"$name header values cannot contain control characters or characters outside Latin-1.")
+    }
+    case _ =>
+  }
+}