diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/implementation/SparkBridgeImplementationInternal.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/implementation/SparkBridgeImplementationInternal.scala index f0cb2d8004da..c947b7801392 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/implementation/SparkBridgeImplementationInternal.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/implementation/SparkBridgeImplementationInternal.scala @@ -17,6 +17,7 @@ import com.fasterxml.jackson.databind.ObjectMapper import scala.collection.convert.ImplicitConversions.`list asScalaBuffer` import scala.collection.mutable +import scala.util.control.NonFatal // scalastyle:off underscore.import import com.azure.cosmos.implementation.feedranges._ @@ -147,6 +148,29 @@ private[cosmos] object SparkBridgeImplementationInternal extends BasicLoggingTra .toArray } + def extractMinLatestLsnFromChangeFeedContinuationOrFallback(continuation: String): Long = { + try { + val continuationTokens = extractContinuationTokensFromChangeFeedStateJson(continuation) + + if (continuationTokens.nonEmpty) { + // FeedRangeContinuation.handleFeedRangeGone expands split ranges by adding child tokens, so the + // minimum across the current token set represents the planned feed range after split handling. + continuationTokens.map(_._2).min + } else { + extractLsnFromChangeFeedContinuation(continuation) + } + } catch { + case NonFatal(rangeEnumerationFailure) => + try { + extractLsnFromChangeFeedContinuation(continuation) + } catch { + case NonFatal(fallbackFailure) => + rangeEnumerationFailure.addSuppressed(fallbackFailure) + throw rangeEnumerationFailure + } + } + } + private[cosmos] def rangeToNormalizedRange(rangeInput: Range[String]) = { val range = FeedRangeInternal.normalizeRange(rangeInput) assert(range != null, "Argument 'range' must not be null.") diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedPartitionReader.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedPartitionReader.scala index 5d83a5139ef2..d8cf4b415bcd 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedPartitionReader.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedPartitionReader.scala @@ -64,7 +64,7 @@ private case class ChangeFeedPartitionReader } private val containerTargetConfig = CosmosContainerConfig.parseCosmosContainerConfig(config) - log.logInfo(s"Reading from feed range ${partition.feedRange}, startLsn $getPartitionStartLsn, " + + log.logInfo(s"Reading from feed range ${partition.feedRange}, startLsn ${startLsn.map(_.toString).getOrElse("n/a")}, " + s"endLsn ${partition.endLsn} of " + s"container ${containerTargetConfig.database}.${containerTargetConfig.container}") private val readConfig = CosmosReadConfig.parseCosmosReadConfig(config) @@ -184,19 +184,17 @@ private case class ChangeFeedPartitionReader } } - private def getPartitionStartLsn: Long = { - if (partition.continuationState.isDefined) { - SparkBridgeImplementationInternal.extractLsnFromChangeFeedContinuation(this.partition.continuationState.get) - } else { - 0 + private def getPartitionStartLsn: Option[Long] = { + partition.continuationState.map { continuationState => + SparkBridgeImplementationInternal.extractMinLatestLsnFromChangeFeedContinuationOrFallback(continuationState) } } private val changeFeedRequestOptions = { - val startLsn = getPartitionStartLsn + val requestStartLsn = startLsn.map(_.toString).getOrElse("n/a") log.logDebug( - s"Request options for Range '${partition.feedRange.min}-${partition.feedRange.max}' LSN '$startLsn'") + s"Request options for Range '${partition.feedRange.min}-${partition.feedRange.max}' LSN '$requestStartLsn'") val options = CosmosChangeFeedRequestOptions .createForProcessingFromContinuation(this.partition.continuationState.get) @@ -263,7 +261,8 @@ private case class ChangeFeedPartitionReader readConfig.maxItemCount, readConfig.prefetchBufferSize, operationContextAndListenerTuple, - this.partition.endLsn + this.partition.endLsn, + startLsn ) override def next(): Boolean = { @@ -294,16 +293,13 @@ private case class ChangeFeedPartitionReader // for cases where the feed range spans multiple physical partitions // pick the smallest lsn Some(SparkBridgeImplementationInternal - .extractContinuationTokensFromChangeFeedStateJson(continuationToken) - .minBy(_._2)._2) + .extractMinLatestLsnFromChangeFeedContinuationOrFallback(continuationToken)) case None => // for change feed, we would only reach here before the first page got fetched // fallback to use the continuation token from the partition instead - Some(SparkBridgeImplementationInternal - .extractContinuationTokensFromChangeFeedStateJson(partition.continuationState.get) - .minBy(_._2)._2) + startLsn } - if (latestLsnOpt.isDefined) latestLsnOpt.get - startLsn else 0 + latestLsnOpt.flatMap(latestLsn => startLsn.map(latestLsn - _)).getOrElse(0) } } diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ItemsPartitionReader.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ItemsPartitionReader.scala index 44027bafbe7e..02a49c1a1441 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ItemsPartitionReader.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ItemsPartitionReader.scala @@ -256,6 +256,7 @@ private case class ItemsPartitionReader readConfig.maxItemCount, readConfig.prefetchBufferSize, operationContextAndListenerTuple, + None, None ) diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIterator.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIterator.scala index c949cd846f6c..c99aebe02f80 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIterator.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIterator.scala @@ -11,7 +11,7 @@ import com.azure.cosmos.util.{CosmosPagedFlux, CosmosPagedIterable} import java.util.concurrent.{ExecutorService, SynchronousQueue, ThreadPoolExecutor, TimeUnit, TimeoutException} import java.util.concurrent.atomic.{AtomicLong, AtomicReference} import scala.util.Random -import scala.util.control.Breaks +import scala.util.control.{Breaks, NonFatal} import scala.concurrent.{Await, ExecutionContext, Future} import com.azure.cosmos.implementation.{ChangeFeedSparkRowItem, OperationCancelledException, SparkBridgeImplementationInternal} @@ -41,7 +41,8 @@ private class TransientIOErrorsRetryingIterator[TSparkRow] val pageSize: Int, val pagePrefetchBufferSize: Int, val operationContextAndListener: Option[OperationContextAndListenerTuple], - val endLsn: Option[Long] + val endLsn: Option[Long], + val startLsn: Option[Long] ) extends BufferedIterator[TSparkRow] with BasicLoggingTrait with AutoCloseable { private[spark] var maxRetryIntervalInMs = CosmosConstants.maxRetryIntervalForTransientFailuresInMs @@ -73,11 +74,7 @@ private class TransientIOErrorsRetryingIterator[TSparkRow] } def getLatestContinuationToken: Option[String] = { - if (lastContinuationToken == null) { - None - } else { - Some(lastContinuationToken.get()) - } + Option(lastContinuationToken.get()) } override def hasNext: Boolean = { @@ -177,6 +174,7 @@ private class TransientIOErrorsRetryingIterator[TSparkRow] None } } else { + validateEofProgressOrThrow() Some(false) } } @@ -237,6 +235,49 @@ private class TransientIOErrorsRetryingIterator[TSparkRow] } } + private[this] def validateEofProgressOrThrow(): Unit = { + endLsn.foreach { targetEndLsn => + val latestMinLsn = try { + Option(lastContinuationToken.get()) + .map(SparkBridgeImplementationInternal.extractMinLatestLsnFromChangeFeedContinuationOrFallback) + } catch { + case NonFatal(parseFailure) => + val message = s"Continuation token parse failure - treating EOF as inconclusive. " + + s"startLsn: ${formatLsn(startLsn)}, endLsn: $targetEndLsn, " + + s"totalChangesCnt: ${totalChangesCnt.get()}, Context: $operationContextString" + val exception = new OperationCancelledException(message, null) + exception.addSuppressed(parseFailure) + logError(message, exception) + throw exception + } + + val isEofValid = latestMinLsn match { + case Some(observedLsn) => observedLsn >= targetEndLsn + case None => startLsn.contains(targetEndLsn) + } + + if (!isEofValid) { + val observedLsnText = latestMinLsn.map(_.toString).getOrElse("no page consumed") + val message = s"Bounded change feed read reached EOF before planned endLsn. " + + s"startLsn: ${formatLsn(startLsn)}, endLsn: $targetEndLsn, " + + s"observed minLatestLsn: $observedLsnText, totalChangesCnt: ${totalChangesCnt.get()}, " + + s"Context: $operationContextString. If this occurred during Spark task cancellation/decommission, " + + s"expect the task to retry from the last committed checkpoint. Continuation tokens are expected " + + s"to preserve split child ranges; range-set shrinkage is undefined behavior." + val exception = new OperationCancelledException(message, null) + + // TODO: Consider moving bounded change-feed EOF validation into a dedicated decorator and short-circuiting + // deterministic zero-progress retries once this policy is separated from transient I/O retry handling. + logError(message, exception) + throw exception + } + } + } + + private[this] def formatLsn(lsn: Option[Long]): String = { + lsn.map(_.toString).getOrElse("n/a") + } + // Clean up iterator references - the underlying Reactor subscription // from Flux.toIterable() will be cleaned up when the iterator is GC'd override def close(): Unit = { diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIteratorITest.scala b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIteratorITest.scala index 1eea43a57daa..a4f1d1e958a3 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIteratorITest.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIteratorITest.scala @@ -113,6 +113,7 @@ class TransientIOErrorsRetryingIteratorITest 2, Queues.XS_BUFFER_SIZE, None, + None, None ) retryingIterator.maxRetryIntervalInMs = 5 @@ -255,6 +256,7 @@ class TransientIOErrorsRetryingIteratorITest 2, Queues.XS_BUFFER_SIZE, None, + None, None ) retryingIterator.maxRetryIntervalInMs = 5 diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIteratorSpec.scala b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIteratorSpec.scala index b8400fdd3eff..dbad259a8c9d 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIteratorSpec.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/TransientIOErrorsRetryingIteratorSpec.scala @@ -3,16 +3,18 @@ package com.azure.cosmos.spark import com.azure.cosmos.CosmosException -import com.azure.cosmos.implementation.SparkRowItem +import com.azure.cosmos.implementation.{ChangeFeedSparkRowItem, OperationCancelledException, SparkRowItem} import com.azure.cosmos.models.{FeedResponse, ModelBridgeInternal} import com.azure.cosmos.spark.diagnostics.BasicLoggingTrait import com.azure.cosmos.util.UtilBridgeInternal import com.fasterxml.jackson.core.JsonProcessingException import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.node.ObjectNode +import org.apache.spark.sql.Row import reactor.core.publisher.Flux import java.time.Duration +import java.util.Base64 import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicLong @@ -45,6 +47,7 @@ class TransientIOErrorsRetryingIteratorSpec extends UnitSpec with BasicLoggingTr pageSize, 1, None, + None, None ) iterator.maxRetryIntervalInMs = 5 @@ -65,6 +68,7 @@ class TransientIOErrorsRetryingIteratorSpec extends UnitSpec with BasicLoggingTr pageSize, 1, None, + None, None ) iterator.maxRetryIntervalInMs = 5 @@ -85,6 +89,7 @@ class TransientIOErrorsRetryingIteratorSpec extends UnitSpec with BasicLoggingTr pageSize, 1, None, + None, None ) iterator.maxRetryIntervalInMs = 5 @@ -108,6 +113,7 @@ class TransientIOErrorsRetryingIteratorSpec extends UnitSpec with BasicLoggingTr pageSize, 1, None, + None, None ) iterator.maxRetryIntervalInMs = 5 @@ -133,6 +139,7 @@ class TransientIOErrorsRetryingIteratorSpec extends UnitSpec with BasicLoggingTr pageSize, 1, None, + None, None ) @@ -165,6 +172,7 @@ class TransientIOErrorsRetryingIteratorSpec extends UnitSpec with BasicLoggingTr pageSize, 1, None, + None, None ) @@ -180,6 +188,311 @@ class TransientIOErrorsRetryingIteratorSpec extends UnitSpec with BasicLoggingTr factoryCallCount.get shouldEqual 1 } + "Bounded change feed reads" should + "not complete when the feed ends before the planned end LSN" in { + + val endLsn = 20L + val lastReturnedLsn = 15L + + val response = generateFeedResponse("ChangeFeed", 1, -1) + ModelBridgeInternal.setFeedResponseContinuationToken( + changeFeedContinuation(lastReturnedLsn), + response + ) + + val iterator = new TransientIOErrorsRetryingIterator( + _ => UtilBridgeInternal.createCosmosPagedFlux( + _ => Flux.fromArray(Array(response)) + ), + pageSize, + 1, + None, + Some(endLsn), + Some(10L) + ) + iterator.maxRetryCount = 0 + + intercept[OperationCancelledException](iterator.hasNext) + } + + "Bounded change feed reads" should + "complete cleanly when the final continuation reaches the planned end LSN" in { + + // Validation matrix row #2: startLsn=10, endLsn=20, single-range continuation=20 -> complete. + val endLsn = 20L + val lastReturnedLsn = 20L + + val response = generateFeedResponse("ChangeFeed", 1, -1) + ModelBridgeInternal.setFeedResponseContinuationToken( + changeFeedContinuation(lastReturnedLsn), + response + ) + + val iterator = new TransientIOErrorsRetryingIterator( + _ => UtilBridgeInternal.createCosmosPagedFlux( + _ => Flux.fromArray(Array(response)) + ), + pageSize, + 1, + None, + Some(endLsn), + Some(10L) + ) + iterator.maxRetryCount = 0 + + iterator.hasNext shouldEqual false + } + + "Bounded change feed reads" should + "complete cleanly when no page is consumed and startLsn already equals endLsn" in { + + // Validation matrix row #3: startLsn=20, endLsn=20, empty flux -> complete. + val endLsn = 20L + + val iterator = new TransientIOErrorsRetryingIterator[SparkRowItem]( + _ => UtilBridgeInternal.createCosmosPagedFlux(_ => Flux.empty()), + pageSize, + 1, + None, + Some(endLsn), + Some(endLsn) + ) + iterator.maxRetryCount = 0 + + iterator.hasNext shouldEqual false + } + + "Bounded change feed reads" should + "throw when no page is consumed and startLsn is below endLsn" in { + + // Validation matrix row #4: startLsn=10, endLsn=20, empty flux -> throws. + val endLsn = 20L + + val iterator = new TransientIOErrorsRetryingIterator[SparkRowItem]( + _ => UtilBridgeInternal.createCosmosPagedFlux(_ => Flux.empty()), + pageSize, + 1, + None, + Some(endLsn), + Some(10L) + ) + iterator.maxRetryCount = 0 + + intercept[OperationCancelledException](iterator.hasNext) + } + + "Bounded change feed reads" should + "throw when any range in a multi-range continuation lags behind the planned end LSN" in { + + // Validation matrix row #5: continuation [20, 18] (min=18) < endLsn=20 -> throws. + val endLsn = 20L + + val response = generateFeedResponse("ChangeFeed", 1, -1) + ModelBridgeInternal.setFeedResponseContinuationToken( + multiRangeChangeFeedContinuation(Seq(20L, 18L)), + response + ) + + val iterator = new TransientIOErrorsRetryingIterator( + _ => UtilBridgeInternal.createCosmosPagedFlux( + _ => Flux.fromArray(Array(response)) + ), + pageSize, + 1, + None, + Some(endLsn), + Some(10L) + ) + iterator.maxRetryCount = 0 + + intercept[OperationCancelledException](iterator.hasNext) + } + + "Unbounded change feed reads" should + "complete cleanly at EOF without any LSN progress validation" in { + + // Validation matrix row #6: endLsn=None (unbounded), empty flux -> no throw. + // Guards against regression of validation kicking in for batch/unbounded mode. + val iterator = new TransientIOErrorsRetryingIterator[SparkRowItem]( + _ => UtilBridgeInternal.createCosmosPagedFlux(_ => Flux.empty()), + pageSize, + 1, + None, + None, + None + ) + iterator.maxRetryCount = 0 + + iterator.hasNext shouldEqual false + } + + "Bounded change feed reads" should + "retry the bounded EOF failure up to maxRetryCount before propagating it" in { + + // Validation matrix row #7: under-run EOF is treated as a transient (408) failure + // and re-subscribes the underlying flux factory until retries are exhausted. + val endLsn = 20L + val lastReturnedLsn = 15L + val maxRetryCount = 2 + val factoryCallCount = new AtomicLong(0) + + val iterator = new TransientIOErrorsRetryingIterator( + _ => { + factoryCallCount.incrementAndGet() + val response = generateFeedResponse("ChangeFeed", 1, -1) + ModelBridgeInternal.setFeedResponseContinuationToken( + changeFeedContinuation(lastReturnedLsn), + response + ) + UtilBridgeInternal.createCosmosPagedFlux( + _ => Flux.fromArray(Array(response)) + ) + }, + pageSize, + 1, + None, + Some(endLsn), + Some(10L) + ) + iterator.maxRetryCount = maxRetryCount + iterator.maxRetryIntervalInMs = 1 + + intercept[OperationCancelledException](iterator.hasNext) + + // 1 initial attempt + maxRetryCount retries + factoryCallCount.get shouldEqual (1 + maxRetryCount) + } + + "Bounded change feed reads" should + "still suppress rows above endLsn while passing the EOF progress check at the boundary" in { + + // Validation matrix row #8: page contains a row with _lsn > endLsn and the + // final continuation reaches endLsn exactly. validateNextLsn must continue to + // suppress the over-LSN row, and validateEofProgressOrThrow must accept the + // boundary continuation without throwing. + val endLsn = 20L + val rowAboveEndLsn = ChangeFeedSparkRowItem(Row.empty, None, "25") + + val response: FeedResponse[ChangeFeedSparkRowItem] = ModelBridgeInternal + .createFeedResponse( + java.util.Collections.singletonList(rowAboveEndLsn), + new ConcurrentHashMap[String, String] + ) + ModelBridgeInternal.setFeedResponseContinuationToken( + changeFeedContinuation(endLsn), + response + ) + + val iterator = new TransientIOErrorsRetryingIterator[ChangeFeedSparkRowItem]( + _ => UtilBridgeInternal.createCosmosPagedFlux( + _ => Flux.fromArray(Array(response)) + ), + pageSize, + 1, + None, + Some(endLsn), + Some(10L) + ) + iterator.maxRetryCount = 0 + + iterator.hasNext shouldEqual false + } + + "Bounded change feed reads" should + "treat malformed continuation EOF as retryable inconclusive progress" in { + + val endLsn = 20L + val response = generateFeedResponse("ChangeFeed", 1, -1) + ModelBridgeInternal.setFeedResponseContinuationToken( + malformedChangeFeedContinuation(), + response + ) + + val iterator = new TransientIOErrorsRetryingIterator( + _ => UtilBridgeInternal.createCosmosPagedFlux( + _ => Flux.fromArray(Array(response)) + ), + pageSize, + 1, + None, + Some(endLsn), + Some(10L) + ) + iterator.maxRetryCount = 0 + + val error = intercept[OperationCancelledException](iterator.hasNext) + error.getStatusCode shouldEqual 408 + error.getMessage should include("Continuation token parse failure") + error.getSuppressed.length should be > 0 + } + + "Bounded change feed reads" should + "complete cleanly when multi-page progress reaches the planned end LSN" in { + + val endLsn = 20L + val rowAtLsn12 = ChangeFeedSparkRowItem(Row.empty, None, "12") + val firstResponse: FeedResponse[ChangeFeedSparkRowItem] = ModelBridgeInternal + .createFeedResponse( + java.util.Collections.singletonList(rowAtLsn12), + new ConcurrentHashMap[String, String] + ) + ModelBridgeInternal.setFeedResponseContinuationToken( + changeFeedContinuation(12L), + firstResponse + ) + + val secondResponse: FeedResponse[ChangeFeedSparkRowItem] = ModelBridgeInternal + .createFeedResponse( + java.util.Collections.emptyList[ChangeFeedSparkRowItem](), + new ConcurrentHashMap[String, String] + ) + ModelBridgeInternal.setFeedResponseContinuationToken( + changeFeedContinuation(endLsn), + secondResponse + ) + + val iterator = new TransientIOErrorsRetryingIterator[ChangeFeedSparkRowItem]( + _ => UtilBridgeInternal.createCosmosPagedFlux( + _ => Flux.fromArray(Array(firstResponse, secondResponse)) + ), + pageSize, + 1, + None, + Some(endLsn), + Some(10L) + ) + iterator.maxRetryCount = 0 + + iterator.hasNext shouldEqual true + iterator.next() shouldEqual rowAtLsn12 + iterator.hasNext shouldEqual false + } + + "Bounded change feed reads" should + "complete cleanly when startLsn equals endLsn and a page continuation reaches the same LSN" in { + + val endLsn = 20L + val response = generateFeedResponse("ChangeFeed", 1, -1) + ModelBridgeInternal.setFeedResponseContinuationToken( + changeFeedContinuation(endLsn), + response + ) + + val iterator = new TransientIOErrorsRetryingIterator( + _ => UtilBridgeInternal.createCosmosPagedFlux( + _ => Flux.fromArray(Array(response)) + ), + pageSize, + 1, + None, + Some(endLsn), + Some(endLsn) + ) + iterator.maxRetryCount = 0 + + iterator.hasNext shouldEqual false + } + private val objectMapper = new ObjectMapper @throws[JsonProcessingException] @@ -340,6 +653,83 @@ class TransientIOErrorsRetryingIteratorSpec extends UnitSpec with BasicLoggingTr } } + private def changeFeedContinuation(lsn: Long): String = { + val state = + s"""{ + | "V": 1, + | "Rid": "testContainer", + | "Mode": "INCREMENTAL", + | "StartFrom": { + | "Type": "BEGINNING" + | }, + | "Continuation": { + | "V": 1, + | "Rid": "testContainer", + | "Continuation": [ + | { + | "token": "$lsn", + | "range": { + | "min": "", + | "max": "FF" + | } + | } + | ], + | "Range": { + | "min": "", + | "max": "FF" + | } + | } + |}""".stripMargin + + Base64.getEncoder.encodeToString(state.getBytes("UTF-8")) + } + + private def malformedChangeFeedContinuation(): String = { + Base64.getEncoder.encodeToString("not-json".getBytes("UTF-8")) + } + + private def multiRangeChangeFeedContinuation(lsns: Seq[Long]): String = { + // Splits the [""..."FF"] range into evenly-sized adjacent sub-ranges, one per supplied LSN. + // The boundaries are arbitrary hex strings; the iterator only inspects the per-range tokens. + require(lsns.nonEmpty, "lsns must contain at least one value") + val boundaries: Seq[String] = "" +: + (1 until lsns.size).map(i => f"${(0xFF * i) / lsns.size}%02X") :+ + "FF" + + val ranges = lsns.zipWithIndex.map { case (lsn, i) => + s"""{ + | "token": "$lsn", + | "range": { + | "min": "${boundaries(i)}", + | "max": "${boundaries(i + 1)}" + | } + |}""".stripMargin + }.mkString(",\n") + + val state = + s"""{ + | "V": 1, + | "Rid": "testContainer", + | "Mode": "INCREMENTAL", + | "StartFrom": { + | "Type": "BEGINNING" + | }, + | "Continuation": { + | "V": 1, + | "Rid": "testContainer", + | "Continuation": [ + | $ranges + | ], + | "Range": { + | "min": "", + | "max": "FF" + | } + | } + |}""".stripMargin + + Base64.getEncoder.encodeToString(state.getBytes("UTF-8")) + } + private class DummyTransientCosmosException extends CosmosException(500, "Dummy Internal Server Error")