[BUG] Cosmos Spark connector can hang indefinitely during change-feed offset discovery when DNS resolution fails

**Affected component:** `com.azure.cosmos.spark:azure-cosmos-spark_3-5_2-12`

#### Problem

The Azure Cosmos DB Spark connector can block indefinitely while discovering change-feed offsets when a Cosmos endpoint cannot be resolved. Instead of failing the streaming query, the connector remains stuck inside a metadata request, preventing Spark and external orchestrators from detecting the failure or retrying.

We encountered this when an Azure-managed private DNS zone group failed to create several regional Cosmos endpoint records. Consequently, requests to the affected regional endpoint produced `UnknownHostException`. Azure documents both the role of private DNS zone groups in maintaining regional records and the relationship between `UnknownHostException` and unresolved Cosmos endpoints. [[Azure Private Link documentation](https://learn.microsoft.com/en-us/azure/cosmos-db/how-to-configure-private-endpoints)](https://learn.microsoft.com/en-us/azure/cosmos-db/how-to-configure-private-endpoints), [[Cosmos Java SDK troubleshooting](https://learn.microsoft.com/en-us/azure/cosmos-db/nosql/troubleshoot-java-sdk-v4)](https://learn.microsoft.com/en-us/azure/cosmos-db/nosql/troubleshoot-java-sdk-v4).

The underlying DNS failure is a separate infrastructure issue. This report concerns the connector’s failure to terminate when that condition persists. (The Azure ticket related to the infra issue is 2607310010001458)

#### Actual behavior

1. A Structured Streaming query using `cosmos.oltp.changeFeed` starts successfully.
2. The first microbatch invokes `ChangeFeedMicroBatchStream.initialOffset()` or `latestOffset()`.
3. Feed-range or partition-metadata discovery enters an untimed Reactor `Mono.block()` call.
4. DNS failures are retried inside the reactive subscription, which may never complete or return control to the connector’s outer retry loop.
5. No microbatch completes, no progress event is emitted, and the streaming query never reports a failure.

In our production incident, applications remained stuck for approximately **2 hours and 45 minutes**, until the scheduler forced termination.

The outer transient-error retry policy also defaults to effectively unlimited attempts, but bounding that policy alone does not solve the problem: an in-flight `Mono.block()` can hang indefinitely without ever returning to the retry loop.

#### Expected behavior

If change-feed metadata discovery cannot complete within a configurable deadline, the connector should:

- Cancel the in-flight metadata request.
- Fail the streaming query with an actionable timeout or underlying connection error.
- Allow Spark and the external orchestrator to retry or reschedule the workload.
- Preserve normal transient-error retries when they complete within the configured deadline.

#### Reproduction

1. Configure a Cosmos DB account with a private endpoint.
2. Make a required regional Cosmos DNS record unavailable, or otherwise cause the selected regional endpoint to consistently return `UnknownHostException`.
3. Start a Spark Structured Streaming query using `cosmos.oltp.changeFeed`.
4. Observe that the streaming query starts but `initialOffset()` or `latestOffset()` never completes, no microbatch makes progress, and no failure is surfaced.

A deterministic unit-test reproduction is to return `Mono.never()` from the metadata request and verify that offset discovery currently blocks indefinitely.

#### Fix approach validated in our fork

We implemented and validated the following mitigation:

1. Introduce `spark.cosmos.changeFeed.maxRetryDurationInSeconds`, defaulting to **300 seconds** and requiring a positive value.
2. Wrap change-feed `initialOffset()` and `latestOffset()` discovery in a shared monotonic deadline.
3. Apply that same deadline across nested metadata operations, retry attempts, and retry backoff; nested operations cannot extend the parent deadline.
4. Replace untimed metadata waits with `Mono.block(remainingDuration)`, ensuring a stalled reactive subscription is canceled when the deadline expires.
5. Propagate the resulting failure to Spark so the streaming query fails and the orchestrator can retry.
6. Scope the deadline to change-feed offset metadata discovery, preserving existing catalog, ordinary read/write, partition-split, and microbatch-processing behavior.

We verified persistent transient failures, nested deadlines, cancellation of a genuinely noncompleting `Mono.never()` request, and interrupted-retry cleanup. Across **440 healthy production samples**, offset discovery had a median of **0.41 seconds**, a p99 of **46 seconds**, and a maximum of **50 seconds**, making the five-minute default conservative.

We can provide the implementation or a patch helpful.