diff --git a/sdk/cosmos/azure-cosmos-spark_3-5/src/main/scala/com/azure/cosmos/spark/ChangeFeedMicroBatchStream.scala b/sdk/cosmos/azure-cosmos-spark_3-5/src/main/scala/com/azure/cosmos/spark/ChangeFeedMicroBatchStream.scala index a2fd36edd61d..e153dbf70139 100644 --- a/sdk/cosmos/azure-cosmos-spark_3-5/src/main/scala/com/azure/cosmos/spark/ChangeFeedMicroBatchStream.scala +++ b/sdk/cosmos/azure-cosmos-spark_3-5/src/main/scala/com/azure/cosmos/spark/ChangeFeedMicroBatchStream.scala @@ -165,6 +165,9 @@ private class ChangeFeedMicroBatchStream log.logDebug(s"--> latestOffset.$streamId") val startChangeFeedOffset = startOffset.asInstanceOf[ChangeFeedOffset] + val operationDeadline = OperationDeadline( + changeFeedConfig.maxRetryDuration, + "Change feed latest offset discovery") val offset = CosmosPartitionPlanner.getLatestOffset( config, startChangeFeedOffset, @@ -176,7 +179,8 @@ private class ChangeFeedMicroBatchStream this.partitioningConfig, this.defaultParallelism, this.container, - Some(this.partitionMetricsMap) + Some(this.partitionMetricsMap), + Some(operationDeadline) ) if (offset.changeFeedState != startChangeFeedOffset.changeFeedState) { @@ -211,7 +215,12 @@ private class ChangeFeedMicroBatchStream assertNotNullOrEmpty(checkpointLocation, "checkpointLocation")) val offsetJson = metadataLog.get(0).getOrElse { val newOffsetJson = CosmosPartitionPlanner.createInitialOffset( - container, containerConfig, changeFeedConfig, partitioningConfig, Some(streamId)) + container, + containerConfig, + changeFeedConfig, + partitioningConfig, + Some(streamId), + Some(OperationDeadline(changeFeedConfig.maxRetryDuration, "Change feed initial offset discovery"))) metadataLog.add(0, newOffsetJson) newOffsetJson } diff --git a/sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md b/sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md index e313b083d215..c462d5c3ff95 100644 --- a/sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md @@ -8,6 +8,8 @@ #### Bugs Fixed +* Added a configurable five-minute deadline for change-feed offset metadata discovery so stalled network operations fail the streaming query instead of blocking indefinitely. See [issue 50021](https://github.com/Azure/azure-sdk-for-java/issues/50021). + #### Other Changes ### 4.49.2 (2026-07-27) diff --git a/sdk/cosmos/azure-cosmos-spark_3-5_2-13/CHANGELOG.md b/sdk/cosmos/azure-cosmos-spark_3-5_2-13/CHANGELOG.md index 395fb7ad6d4d..3fc2b020a6f5 100644 --- a/sdk/cosmos/azure-cosmos-spark_3-5_2-13/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-spark_3-5_2-13/CHANGELOG.md @@ -8,6 +8,8 @@ #### Bugs Fixed +* Added a configurable five-minute deadline for change-feed offset metadata discovery so stalled network operations fail the streaming query instead of blocking indefinitely. See [issue 50021](https://github.com/Azure/azure-sdk-for-java/issues/50021). + #### Other Changes ### 4.49.2 (2026-07-27) diff --git a/sdk/cosmos/azure-cosmos-spark_3/docs/configuration-reference.md b/sdk/cosmos/azure-cosmos-spark_3/docs/configuration-reference.md index 53c670fb208a..ac1fdf60e8c5 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/docs/configuration-reference.md +++ b/sdk/cosmos/azure-cosmos-spark_3/docs/configuration-reference.md @@ -122,6 +122,7 @@ Used to influence the json serialization/deserialization behavior | `spark.cosmos.changeFeed.itemCountPerTriggerHint` | None (process all available data in first micro-batch) | Approximate maximum number of items read from change feed for each micro-batch/trigger. If not set, all available data in the changefeed is going to be processed in the first micro-batch. This could overload the client-resources (especially memory), so choosing a value to cap the resource consumption in the Spark executors is advisable here. Usually a reasonable value would be at least in the 100-thousands or single-digit millions. | | `spark.cosmos.changeFeed.batchCheckpointLocation` | None | Can be used to generate checkpoints when using change feed queries in batch mode - and proceeding on the next iteration where the previous left off. | | `spark.cosmos.changeFeed.performance.monitoring.enabled` | `true` | A Flag to indicate whether enable change feed performance monitoring. When enabled, custom task metrics will be tracked internally, which will be used to dynamically tuning the change feed micro-batch size. | +| `spark.cosmos.changeFeed.maxRetryDurationInSeconds` | `300` | Maximum duration in seconds allowed for change-feed offset metadata discovery (`initialOffset` and `latestOffset`) in the Spark 3.5 connector. 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. | #### Json conversion configuration | Config Property Name | Default | Description | diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosConfig.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosConfig.scala index 3e6461b94c89..39708c8c28b4 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosConfig.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosConfig.scala @@ -143,6 +143,7 @@ private[spark] object CosmosConfigNames { val ChangeFeedBatchCheckpointLocation = "spark.cosmos.changeFeed.batchCheckpointLocation" val ChangeFeedBatchCheckpointLocationIgnoreWhenInvalid = "spark.cosmos.changeFeed.batchCheckpointLocation.ignoreWhenInvalid" val ChangeFeedPerformanceMonitoringEnabled = "spark.cosmos.changeFeed.performance.monitoring.enabled" + val ChangeFeedMaxRetryDurationInSeconds = "spark.cosmos.changeFeed.maxRetryDurationInSeconds" val ThroughputControlEnabled = "spark.cosmos.throughputControl.enabled" val ThroughputControlAccountEndpoint = "spark.cosmos.throughputControl.accountEndpoint" val ThroughputControlAccountKey = "spark.cosmos.throughputControl.accountKey" @@ -281,6 +282,7 @@ private[spark] object CosmosConfigNames { ChangeFeedBatchCheckpointLocation, ChangeFeedBatchCheckpointLocationIgnoreWhenInvalid, ChangeFeedPerformanceMonitoringEnabled, + ChangeFeedMaxRetryDurationInSeconds, ThroughputControlEnabled, ThroughputControlAccountEndpoint, ThroughputControlAccountKey, @@ -2370,7 +2372,8 @@ private case class CosmosChangeFeedConfig maxItemCountPerTrigger: Option[Long], batchCheckpointLocation: Option[String], ignoreOffsetWhenInvalid: Boolean, - performanceMonitoringEnabled: Boolean + performanceMonitoringEnabled: Boolean, + maxRetryDuration: Duration ) { def toRequestOptions(feedRange: FeedRange): CosmosChangeFeedRequestOptions = { @@ -2402,6 +2405,7 @@ private object CosmosChangeFeedConfig { private val DefaultStartFromMode: ChangeFeedStartFromMode = ChangeFeedStartFromModes.Beginning private val DefaultIgnoreOffsetWhenInvalid: Boolean = false private val DefaultPerformanceMonitoringEnabled: Boolean = true + private val DefaultMaxRetryDuration: Duration = Duration.ofSeconds(300) private val startFrom = CosmosConfigEntry[ChangeFeedStartFromMode]( key = CosmosConfigNames.ChangeFeedStartFrom, @@ -2459,6 +2463,18 @@ private object CosmosChangeFeedConfig { helpMessage = "A Flag to indicate whether enable change feed performance monitoring." + " When enabled, custom task metrics will be tracked internally, which will be used to dynamically tuning the change feed micro-batch size.") + private val maxRetryDuration = CosmosConfigEntry[Duration]( + key = CosmosConfigNames.ChangeFeedMaxRetryDurationInSeconds, + mandatory = false, + defaultValue = Some(DefaultMaxRetryDuration), + parseFromStringFunction = value => { + val duration = Duration.ofSeconds(value.toLong) + require(!duration.isZero && !duration.isNegative, "value must be greater than zero") + duration + }, + helpMessage = "Maximum duration in seconds allowed for change feed offset metadata discovery. " + + "The value must be greater than zero.") + private def validateStartFromMode(startFrom: String): ChangeFeedStartFromMode = { Option(startFrom).fold(DefaultStartFromMode)(sf => { val trimmed = sf.trim @@ -2484,6 +2500,7 @@ private object CosmosChangeFeedConfig { } val batchCheckpointLocationParsed = CosmosConfigEntry.parse(cfg, batchCheckpointLocation) val performanceMonitoringEnabledParsed = CosmosConfigEntry.parse(cfg, performanceMonitoringEnabled) + val maxRetryDurationParsed = CosmosConfigEntry.parse(cfg, maxRetryDuration) CosmosChangeFeedConfig( changeFeedModeParsed.getOrElse(DefaultChangeFeedMode), @@ -2492,7 +2509,8 @@ private object CosmosChangeFeedConfig { maxItemCountPerTriggerHintParsed, batchCheckpointLocationParsed, ignoreOffsetWhenInvalidParsed.getOrElse(DefaultIgnoreOffsetWhenInvalid), - performanceMonitoringEnabledParsed.get + performanceMonitoringEnabledParsed.get, + maxRetryDurationParsed.get ) } } diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosPartitionPlanner.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosPartitionPlanner.scala index 9b7ffbf1cbd9..74b88ed3df37 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosPartitionPlanner.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosPartitionPlanner.scala @@ -41,7 +41,8 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { defaultMaxPartitionSizeInMB: Int, readLimit: ReadLimit, isChangeFeed: Boolean, - partitionMetricsMap: Option[ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]] = None + partitionMetricsMap: Option[ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]] = None, + operationDeadline: Option[OperationDeadline] = None ): Array[CosmosInputPartition] = { TransientErrorsRetryPolicy.executeWithRetry(() => @@ -53,7 +54,8 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { defaultMaxPartitionSizeInMB, readLimit, isChangeFeed, - partitionMetricsMap)) + partitionMetricsMap), + operationDeadline = operationDeadline) } private[this] def createInputPartitionsImpl @@ -171,12 +173,14 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { containerConfig: CosmosContainerConfig, changeFeedConfig: CosmosChangeFeedConfig, partitioningConfig: CosmosPartitioningConfig, - streamId: Option[String] + streamId: Option[String], + operationDeadline: Option[OperationDeadline] = None ): String = { TransientErrorsRetryPolicy.executeWithRetry(() => - createInitialOffsetImpl(container, containerConfig, changeFeedConfig, partitioningConfig, streamId) - ) + createInitialOffsetImpl( + container, containerConfig, changeFeedConfig, partitioningConfig, streamId, operationDeadline), + operationDeadline = operationDeadline) } // scalastyle:off method.length @@ -186,14 +190,15 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { containerConfig: CosmosContainerConfig, changeFeedConfig: CosmosChangeFeedConfig, partitioningConfig: CosmosPartitioningConfig, - streamId: Option[String] + streamId: Option[String], + operationDeadline: Option[OperationDeadline] ): String = { assertOnSparkDriver() val lastContinuationTokens: ConcurrentMap[FeedRange, String] = new ConcurrentHashMap[FeedRange, String]() val shouldRefreshFeedRangesInCache = new AtomicBoolean(false) // only need to refresh the cache if a partition split has been detected - ContainerFeedRangesCache + val initialOffsets = ContainerFeedRangesCache .getFeedRanges(container, containerConfig.feedRangeRefreshIntervalInSecondsOpt) .map(feedRangeList => partitioningConfig.feedRangeFiler match { @@ -209,7 +214,7 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { .flatMap(feedRange => { TransientErrorsRetryPolicy.executeWithRetry(() => { queryChangeFeedForInitialOffset(changeFeedConfig, feedRange, container) - }) + }, operationDeadline = operationDeadline) }) .doOnNext(feedRangeContinuationMap => { if (feedRangeContinuationMap.size > 1) { @@ -230,7 +235,11 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { }) .asJava() .collectList() - .block() + + operationDeadline match { + case Some(deadline) => deadline.block(initialOffsets) + case None => initialOffsets.block() + } if (shouldRefreshFeedRangesInCache.get()) { logDebug("Feed range split has been detected, forcing refresh of the feed ranges in cache") @@ -339,7 +348,8 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { partitioningConfig: CosmosPartitioningConfig, defaultParallelism: Int, container: CosmosAsyncContainer, - partitionMetricsMap: Option[ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]] = None + partitionMetricsMap: Option[ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]] = None, + operationDeadline: Option[OperationDeadline] = None ): ChangeFeedOffset = { TransientErrorsRetryPolicy.executeWithRetry(() => @@ -354,7 +364,9 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { partitioningConfig, defaultParallelism, container, - partitionMetricsMap)) + partitionMetricsMap, + operationDeadline), + operationDeadline = operationDeadline) } // scalastyle:on parameter.number @@ -373,7 +385,8 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { partitioningConfig: CosmosPartitioningConfig, defaultParallelism: Int, container: CosmosAsyncContainer, - partitionMetricsMap: Option[ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]] = None + partitionMetricsMap: Option[ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]] = None, + operationDeadline: Option[OperationDeadline] ): ChangeFeedOffset = { assertOnSparkDriver() assertNotNull(startOffset, "startOffset") @@ -385,7 +398,8 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { containerConfig, partitioningConfig, true, - Some(maxStaleness) + Some(maxStaleness), + operationDeadline ) val defaultMaxPartitionSizeInMB = DefaultPartitionSizeInMB @@ -402,7 +416,8 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { defaultMaxPartitionSizeInMB, readLimit, true, - partitionMetricsMap + partitionMetricsMap, + operationDeadline ) if (isDebugLogEnabled) { @@ -771,7 +786,8 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { cosmosContainerConfig: CosmosContainerConfig, partitionConfig: CosmosPartitioningConfig, isChangeFeed: Boolean, - maxStaleness: Option[Duration] = None + maxStaleness: Option[Duration] = None, + operationDeadline: Option[OperationDeadline] = None ): Array[PartitionMetadata] = { TransientErrorsRetryPolicy.executeWithRetry(() => @@ -782,7 +798,9 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { cosmosContainerConfig, partitionConfig, isChangeFeed, - maxStaleness)) + maxStaleness, + operationDeadline), + operationDeadline = operationDeadline) } private[this] def getPartitionMetadataImpl( @@ -792,11 +810,12 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { cosmosContainerConfig: CosmosContainerConfig, partitionConfig: CosmosPartitioningConfig, isChangeFeed: Boolean, - maxStaleness: Option[Duration] = None + maxStaleness: Option[Duration] = None, + operationDeadline: Option[OperationDeadline] = None ): Array[PartitionMetadata] = { assertOnSparkDriver() - this + val partitionMetadata = this .getFeedRanges( userConfig, cosmosClientConfig, @@ -852,8 +871,10 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait { expandPartitionMetadataByLatestLsn(metadata, isChangeFeed), partitionConfig.feedRangeFiler)) }) - .block() - .toArray + (operationDeadline match { + case Some(deadline) => deadline.block(partitionMetadata.asJava()) + case None => partitionMetadata.block() + }).toArray } private[spark] def expandPartitionMetadataByLatestLsn( diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/OperationDeadline.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/OperationDeadline.scala new file mode 100644 index 000000000000..31954d44679d --- /dev/null +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/OperationDeadline.scala @@ -0,0 +1,89 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.spark + +import reactor.core.{Exceptions => ReactorExceptions} +import reactor.core.publisher.Mono + +import java.time.Duration +import java.util.concurrent.TimeoutException + +private[spark] final class OperationDeadline private( + timeout: Duration, + operationName: String, + deadlineInNanos: Long) { + + def remainingDuration: Duration = { + if (deadlineInNanos == Long.MaxValue) { + Duration.ofNanos(Long.MaxValue) + } else { + val remainingNanos = deadlineInNanos - System.nanoTime() + if (remainingNanos <= 0) { + throw timeoutException() + } + + Duration.ofNanos(remainingNanos) + } + } + + def block[T](mono: Mono[T]): T = { + try { + mono.timeout(remainingDuration).block() + } catch { + case error: RuntimeException if ReactorExceptions.unwrap(error).isInstanceOf[TimeoutException] => + throw timeoutException(error) + } + } + + def sleep(durationInMillis: Int): Unit = { + val remaining = remainingDuration + val requested = Duration.ofMillis(durationInMillis.toLong) + val effective = if (requested.compareTo(remaining) < 0) requested else remaining + + val millis = effective.toMillis + val additionalNanos = (effective.minusMillis(millis).toNanos).toInt + try { + Thread.sleep(millis, additionalNanos) + } catch { + case interrupted: InterruptedException => + Thread.currentThread().interrupt() + throw interrupted + } + + if (requested.compareTo(remaining) >= 0) { + remainingDuration + } + } + + private def timeoutException(cause: Throwable = null): TimeoutException = { + val exception = new TimeoutException( + s"$operationName did not complete within $timeout.") + if (cause != null) { + exception.initCause(cause) + } + exception + } +} + +private[spark] object OperationDeadline { + def apply(timeout: Duration, operationName: String): OperationDeadline = { + require(timeout != null && !timeout.isZero && !timeout.isNegative, "timeout must be positive") + + val timeoutNanos = try { + timeout.toNanos + } catch { + case _: ArithmeticException => Long.MaxValue + } + val deadline = if (timeoutNanos == Long.MaxValue) { + Long.MaxValue + } else { + try { + Math.addExact(System.nanoTime(), timeoutNanos) + } catch { + case _: ArithmeticException => Long.MaxValue + } + } + new OperationDeadline(timeout, operationName, deadline) + } +} diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransientErrorsRetryPolicy.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransientErrorsRetryPolicy.scala index 7218e83df7f1..4c360ca98a3a 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransientErrorsRetryPolicy.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransientErrorsRetryPolicy.scala @@ -19,7 +19,8 @@ private[spark] object TransientErrorsRetryPolicy extends BasicLoggingTrait { initialMaxRetryIntervalInMs: Int = CosmosConstants.initialMaxRetryIntervalForTransientFailuresInMs, maxRetryIntervalInMs: Int = CosmosConstants.maxRetryIntervalForTransientFailuresInMs, maxRetryCount: Int = Int.MaxValue, - statusResetFuncBetweenRetry: Option[() => Unit] = None + statusResetFuncBetweenRetry: Option[() => Unit] = None, + operationDeadline: Option[OperationDeadline] = None ): T = { val loop = new Breaks() val retryCount = new AtomicLong(0) @@ -31,6 +32,7 @@ private[spark] object TransientErrorsRetryPolicy extends BasicLoggingTrait { val retryIntervalInMs = rnd.nextInt(currentMaxRetryIntervalInMs) try { + operationDeadline.foreach(_.remainingDuration) returnValue = Some(func()) loop.break } @@ -60,7 +62,17 @@ private[spark] object TransientErrorsRetryPolicy extends BasicLoggingTrait { statusResetFuncBetweenRetry.get.apply() } - Thread.sleep(retryIntervalInMs) + operationDeadline match { + case Some(deadline) => deadline.sleep(retryIntervalInMs) + case None => + try { + Thread.sleep(retryIntervalInMs) + } catch { + case interrupted: InterruptedException => + Thread.currentThread().interrupt() + throw interrupted + } + } currentMaxRetryIntervalInMs = Math.min(2 * currentMaxRetryIntervalInMs, maxRetryIntervalInMs) } } diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/CosmosConfigSpec.scala b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/CosmosConfigSpec.scala index f9281d106362..a691f7f07acf 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/CosmosConfigSpec.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/CosmosConfigSpec.scala @@ -948,6 +948,7 @@ class CosmosConfigSpec extends UnitSpec with BasicLoggingTrait { config.startFrom shouldEqual ChangeFeedStartFromModes.Beginning config.startFromPointInTime shouldEqual None config.maxItemCountPerTrigger shouldEqual None + config.maxRetryDuration shouldEqual Duration.ofSeconds(300) } it should "parse change feed config for full fidelity with incorrect casing" in { @@ -1025,6 +1026,22 @@ class CosmosConfigSpec extends UnitSpec with BasicLoggingTrait { config.performanceMonitoringEnabled shouldBe false } + it should "parse change feed max retry duration" in { + val config = CosmosChangeFeedConfig.parseCosmosChangeFeedConfig(Map( + CosmosConfigNames.ChangeFeedMaxRetryDurationInSeconds -> "42")) + + config.maxRetryDuration shouldEqual Duration.ofSeconds(42) + } + + it should "reject a non-positive change feed max retry duration" in { + val exception = intercept[RuntimeException] { + CosmosChangeFeedConfig.parseCosmosChangeFeedConfig(Map( + CosmosConfigNames.ChangeFeedMaxRetryDurationInSeconds -> "0")) + } + + exception.getMessage should include (CosmosConfigNames.ChangeFeedMaxRetryDurationInSeconds) + } + it should "complain when parsing invalid change feed mode" in { val changeFeedConfig = Map( "spark.cosmos.changeFeed.mode" -> "Whatever", diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/OperationDeadlineSpec.scala b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/OperationDeadlineSpec.scala new file mode 100644 index 000000000000..4a5a84057f1e --- /dev/null +++ b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/OperationDeadlineSpec.scala @@ -0,0 +1,52 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.spark + +import reactor.core.publisher.Mono + +import java.time.Duration +import java.util.concurrent.TimeoutException +import java.util.concurrent.atomic.AtomicBoolean + +class OperationDeadlineSpec extends UnitSpec { + "OperationDeadline" should "cancel a non-completing reactive operation when the deadline expires" in { + val cancelled = new AtomicBoolean(false) + val neverCompletingOperation = Mono.never[String]().doOnCancel(() => cancelled.set(true)) + val deadline = OperationDeadline(Duration.ofMillis(25), "Test metadata discovery") + + val exception = intercept[TimeoutException] { + deadline.block(neverCompletingOperation) + } + + exception.getMessage should include ("Test metadata discovery") + exception.getMessage should include ("PT0.025S") + cancelled.get shouldBe true + } + + it should "reject a non-positive timeout" in { + intercept[IllegalArgumentException] { + OperationDeadline(Duration.ZERO, "Test metadata discovery") + } + } + + it should "preserve the interrupt status when sleep is interrupted" in { + val deadline = OperationDeadline(Duration.ofSeconds(1), "Test metadata discovery") + Thread.currentThread().interrupt() + + try { + intercept[InterruptedException] { + deadline.sleep(1) + } + Thread.currentThread().isInterrupted shouldBe true + } finally { + Thread.interrupted() + } + } + + it should "support timeout durations that cannot be represented in nanoseconds" in { + val deadline = OperationDeadline(Duration.ofSeconds(Long.MaxValue), "Test metadata discovery") + + deadline.remainingDuration shouldEqual Duration.ofNanos(Long.MaxValue) + } +} diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientErrorsRetryPolicySpec.scala b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientErrorsRetryPolicySpec.scala index aa803e089d31..84f1e8aa5beb 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientErrorsRetryPolicySpec.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientErrorsRetryPolicySpec.scala @@ -81,6 +81,22 @@ class TransientErrorsRetryPolicySpec extends UnitSpec with BasicLoggingTrait { thrownException.get.getStatusCode shouldEqual new DummyTransientCosmosException().getStatusCode } + "TransientErrorsRetryPolicy" should "preserve the interrupt status when retry backoff is interrupted" in { + Thread.currentThread().interrupt() + + try { + intercept[InterruptedException] { + TransientErrorsRetryPolicy.executeWithRetry( + () => throw new DummyTransientCosmosException(), + initialMaxRetryIntervalInMs = 1, + maxRetryIntervalInMs = 1) + } + Thread.currentThread().isInterrupted shouldBe true + } finally { + Thread.interrupted() + } + } + private class DummyTransientCosmosException extends CosmosException(500, "Dummy Internal Server Error")