Fix Cosmos Spark change feed offset discovery timeout

[Arnab Nandy (arnabnandy7)]

Description

Fixes Azure Cosmos DB Spark change feed streaming queries potentially blocking indefinitely during offset metadata discovery when an endpoint cannot be resolved or another metadata request does not complete.

This pull request:

• Adds the spark.cosmos.changeFeed.maxRetryDurationInSeconds configuration option, with a default of 300 seconds.
• Validates that the configured duration is greater than zero.
• Applies a shared monotonic deadline to initialOffset() and latestOffset() metadata discovery.
• Propagates the deadline across nested metadata operations, transient-error retries, and retry backoff.
• Cancels non-completing reactive metadata requests when the deadline expires.
• Fails the streaming query with an actionable timeout instead of allowing it to block indefinitely.
• Preserves existing behavior for batch change feed operations, catalog operations, ordinary reads and writes, and micro-batch processing.
• Adds unit coverage for configuration parsing and cancellation of a non-completing reactive request.
• Updates the Spark 3.5 connector changelogs and configuration reference.

Fixes #50021

All SDK Contribution checklist:

• The pull request does not introduce [breaking changes]
• CHANGELOG is updated for new features, bug fixes or other significant changes.
• I have read the contribution guidelines.

General Guidelines and Best Practices

• Title of the pull request is clear and informative.
• There are a small number of commits, each of which have an informative message. This means that previously merged commits do not appear in the history of the PR. For more information on cleaning up the commits in your PR, see this page.

Testing Guidelines

• Pull request includes test coverage for the included changes.

[github-actions]

Thank you for your contribution Arnab Nandy (@arnabnandy7)! We will review the pull request and get back to you soon.

[azure-pipelines]

Azure Pipelines:
Successfully started running 2 pipeline(s).
32 pipeline(s) were filtered out due to trigger conditions.
There may be pipelines that require an authorized user to comment /azp run to run.

[Copilot]

Pull request overview

This PR addresses a reliability issue in the Cosmos DB Spark Structured Streaming change feed connector where initialOffset() / latestOffset() metadata discovery could block indefinitely (e.g., during DNS resolution failures). It introduces a configurable deadline and propagates it through nested metadata calls and transient retry/backoff, ensuring stalled operations are cancelled and surfaced as actionable timeouts.

Changes:

• Added spark.cosmos.changeFeed.maxRetryDurationInSeconds (default 300s) with validation and documentation updates.
• Introduced OperationDeadline and threaded it through change-feed offset discovery, including reactive block(...) and retry backoff sleeps.
• Added unit tests for config parsing and for cancelling a non-completing reactive operation on deadline expiry.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/OperationDeadline.scala	New deadline utility used to bound blocking waits and backoff with a monotonic deadline.
sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransientErrorsRetryPolicy.scala	Added optional OperationDeadline support to stop retries/backoff once the deadline expires.
sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosPartitionPlanner.scala	Propagated the shared deadline through nested offset/metadata discovery and blocking points.
sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosConfig.scala	Added the new config key and parsed it into CosmosChangeFeedConfig.
sdk/cosmos/azure-cosmos-spark_3-5/src/main/scala/com/azure/cosmos/spark/ChangeFeedMicroBatchStream.scala	Applied the deadline to initialOffset() and latestOffset() discovery in Spark 3.5 streaming.
sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/OperationDeadlineSpec.scala	New unit test validating cancellation and input validation for OperationDeadline.
sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/CosmosConfigSpec.scala	Added unit coverage for default/parsed/invalid max retry duration config.
sdk/cosmos/azure-cosmos-spark_3/docs/configuration-reference.md	Documented the new configuration option and semantics.
sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md	Documented the bug fix in the Spark 3.5 (Scala 2.12) connector changelog.
sdk/cosmos/azure-cosmos-spark_3-5_2-13/CHANGELOG.md	Documented the bug fix in the Spark 3.5 (Scala 2.13) connector changelog.

[Copilot]

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated no new comments.

Suppressed comments (1)

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/OperationDeadline.scala:34

• OperationDeadline.block detects a Reactor timeout by matching the prefix of an IllegalStateException message ("Timeout on blocking read..."). This is brittle across Reactor versions/localization and can cause timeouts to surface as a less-actionable IllegalStateException instead of the intended TimeoutException with the operation name/deadline context. Prefer using Mono.timeout(remaining) and unwrap the thrown exception to reliably detect timeouts without string matching.

      case timeoutOnBlockingRead: IllegalStateException
        if timeoutOnBlockingRead.getMessage != null &&
          timeoutOnBlockingRead.getMessage.startsWith("Timeout on blocking read") =>
        throw timeoutException(timeoutOnBlockingRead)
    }

[Copilot]

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated no new comments.

Suppressed comments (1)

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/OperationDeadline.scala:33

• OperationDeadline.block detects Reactor blocking timeouts by checking IllegalStateException message with startsWith("Timeout on blocking read"). This is brittle (message formatting can vary) and is inconsistent with other usages in this repo that match via contains. Switching to contains reduces the risk of missing a timeout and leaking the less actionable IllegalStateException to callers.

      case timeoutOnBlockingRead: IllegalStateException
        if timeoutOnBlockingRead.getMessage != null &&
          timeoutOnBlockingRead.getMessage.startsWith("Timeout on blocking read") =>
        throw timeoutException(timeoutOnBlockingRead)
    }

[Kevin Pilch (Pilchie)]

Annie Liang (@xinlian12) - can you take a look at this?

[Copilot]

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated no new comments.

Suppressed comments (1)

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/OperationDeadline.scala:46

• OperationDeadline.sleep unconditionally throws TimeoutException when the requested sleep duration is >= the remaining budget, even though Thread.sleep can return before the deadline actually elapses (e.g., due to timer granularity). This can cause premature timeouts during retry backoff.

Consider re-checking the deadline after sleeping and only throwing if it has actually expired.

    if (requested.compareTo(remaining) >= 0) {
      throw timeoutException()
    }

[Copilot]

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated no new comments.

Suppressed comments (1)

sdk/cosmos/azure-cosmos-spark_3/docs/configuration-reference.md:125

• The configuration reference documents spark.cosmos.changeFeed.maxRetryDurationInSeconds as applying to change-feed offset discovery in general, but the deadline is only wired up in the Spark 3.5 streaming implementation (azure-cosmos-spark_3-5); the Spark 3.3/3.4 connectors' ChangeFeedMicroBatchStream still call CosmosPartitionPlanner without an OperationDeadline. This doc entry should clarify the current scope to avoid users of other Spark 3.x artifacts assuming it will prevent hangs.

| `spark.cosmos.changeFeed.maxRetryDurationInSeconds`         | `300`                                                  | Maximum duration in seconds allowed for change-feed offset metadata discovery (`initialOffset` and `latestOffset`). The deadline includes metadata requests, transient-error retries, and retry backoff. When the deadline expires, the streaming query fails so Spark or an external orchestrator can retry it. The value must be greater than zero.                                                                                                                                                         |

[Copilot]

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated no new comments.

Suppressed comments (2)

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransientErrorsRetryPolicy.scala:68

• Thread.sleep clears the thread’s interrupted status when it throws InterruptedException. Since this method doesn’t restore the interrupt flag, callers higher up may not be able to detect the interrupt. Catch InterruptedException, re-interrupt the thread, and rethrow to preserve expected interruption semantics during retry backoff (including when OperationDeadline.sleep is used).

        operationDeadline match {
          case Some(deadline) => deadline.sleep(retryIntervalInMs)
          case None => Thread.sleep(retryIntervalInMs)
        }

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/OperationDeadline.scala:46

• Thread.sleep clears the interrupted status when throwing InterruptedException. Re-interrupting the thread before rethrowing avoids losing the interrupt signal (which can lead to harder-to-debug shutdown/cancellation behavior when Spark is stopping a query).

    Thread.sleep(millis, additionalNanos)

    if (requested.compareTo(remaining) >= 0) {
      remainingDuration
    }

[Copilot]

Pull request overview

Copilot reviewed 11 out of 11 changed files in this pull request and generated no new comments.

Suppressed comments (1)

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/OperationDeadline.scala:24

• remainingDuration can overflow and become negative when deadlineInNanos is Long.MaxValue (used when the configured timeout can’t be represented in nanos) and System.nanoTime() is negative, causing an immediate timeout even though the deadline is intended to be effectively unbounded. Consider special-casing Long.MaxValue to avoid subtracting nanoTime() in that case.

  def remainingDuration: Duration = {
    val remainingNanos = deadlineInNanos - System.nanoTime()
    if (remainingNanos <= 0) {
      throw timeoutException()
    }

[Copilot]

Pull request overview

Copilot reviewed 11 out of 11 changed files in this pull request and generated no new comments.