support feedrange cache refresh interval config (#46759)

* support feedrange cache refresh interval config

---------

Co-authored-by: annie-mac <xinlian@microsoft.com>

Author: xinlian12

sdk/cosmos/azure-cosmos-spark_3-3_2-12/CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 4.40.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added support for feed range cache refresh interval config. - See [46759](https://github.com/Azure/azure-sdk-for-java/pull/46759)
 
 #### Breaking Changes
 

sdk/cosmos/azure-cosmos-spark_3-3_2-12/src/main/scala/com/azure/cosmos/spark/ChangeFeedMicroBatchStream.scala

@@ -210,7 +210,7 @@ private class ChangeFeedMicroBatchStream
         assertNotNullOrEmpty(checkpointLocation, "checkpointLocation"))
     val offsetJson = metadataLog.get(0).getOrElse {
       val newOffsetJson = CosmosPartitionPlanner.createInitialOffset(
-        container, changeFeedConfig, partitioningConfig, Some(streamId))
+        container, containerConfig, changeFeedConfig, partitioningConfig, Some(streamId))
       metadataLog.add(0, newOffsetJson)
       newOffsetJson
     }

sdk/cosmos/azure-cosmos-spark_3-4_2-12/CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 4.40.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added support for feed range cache refresh interval config. - See [46759](https://github.com/Azure/azure-sdk-for-java/pull/46759)
 
 #### Breaking Changes
 

sdk/cosmos/azure-cosmos-spark_3-4_2-12/src/main/scala/com/azure/cosmos/spark/ChangeFeedMicroBatchStream.scala

@@ -211,7 +211,7 @@ private class ChangeFeedMicroBatchStream
         assertNotNullOrEmpty(checkpointLocation, "checkpointLocation"))
     val offsetJson = metadataLog.get(0).getOrElse {
       val newOffsetJson = CosmosPartitionPlanner.createInitialOffset(
-        container, changeFeedConfig, partitioningConfig, Some(streamId))
+        container, containerConfig, changeFeedConfig, partitioningConfig, Some(streamId))
       metadataLog.add(0, newOffsetJson)
       newOffsetJson
     }

sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 4.40.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added support for feed range cache refresh interval config. - See [46759](https://github.com/Azure/azure-sdk-for-java/pull/46759)
 
 #### Breaking Changes
 

sdk/cosmos/azure-cosmos-spark_3-5_2-12/src/main/scala/com/azure/cosmos/spark/ChangeFeedMicroBatchStream.scala

@@ -212,7 +212,7 @@ private class ChangeFeedMicroBatchStream
         assertNotNullOrEmpty(checkpointLocation, "checkpointLocation"))
     val offsetJson = metadataLog.get(0).getOrElse {
       val newOffsetJson = CosmosPartitionPlanner.createInitialOffset(
-        container, changeFeedConfig, partitioningConfig, Some(streamId))
+        container, containerConfig, changeFeedConfig, partitioningConfig, Some(streamId))
       metadataLog.add(0, newOffsetJson)
       newOffsetJson
     }

sdk/cosmos/azure-cosmos-spark_3_2-12/docs/configuration-reference.md

@@ -50,6 +50,7 @@ import scala.collection.JavaConverters._
 private class BulkWriter
 (
   container: CosmosAsyncContainer,
+  containerConfig: CosmosContainerConfig,
   partitionKeyDefinition: PartitionKeyDefinition,
   writeConfig: CosmosWriteConfig,
   diagnosticsConfig: DiagnosticsConfig,
@@ -74,7 +75,7 @@ private class BulkWriter
     case Some(configuredMaxConcurrentPartitions) => 2 * configuredMaxConcurrentPartitions
     // using the total number of physical partitions
     // multiplied by 2 to leave space for partition splits during ingestion
-    case None => 2 * ContainerFeedRangesCache.getFeedRanges(container).block().size
+    case None => 2 * ContainerFeedRangesCache.getFeedRanges(container, containerConfig.feedRangeRefreshIntervalInSecondsOpt).block().size
   }
   log.logInfo(
     s"BulkWriter instantiated (Host CPU count: $cpuCount, maxPendingOperations: $maxPendingOperations, " +

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/spark/BulkWriter.scala

@@ -99,7 +99,7 @@ private class ChangeFeedBatch
               SparkBridgeImplementationInternal.extractCollectionRid(changeFeedStateBase64))
 
             val newOffsetJson = CosmosPartitionPlanner.createInitialOffset(
-              container, changeFeedConfig, partitioningConfig, None)
+              container, containerConfig, changeFeedConfig, partitioningConfig, None)
             log.logWarning(s"Invalid Start offset '$offsetJson' retrieved from file location " +
               s"'$startOffsetLocation' for batchId: $batchId -> New offset retrieved from " +
               s"service: '$newOffsetJson'.")
@@ -113,14 +113,14 @@ private class ChangeFeedBatch
             newOffsetJson
           }
         } else {
-          val newOffsetJson = CosmosPartitionPlanner.createInitialOffset(container, changeFeedConfig, partitioningConfig, None)
+          val newOffsetJson = CosmosPartitionPlanner.createInitialOffset(container, containerConfig, changeFeedConfig, partitioningConfig, None)
           log.logDebug(s"No Start offset retrieved from file location '$startOffsetLocation' for batchId: $batchId " +
             s"-> offset retrieved from service: '$newOffsetJson'.")
 
           newOffsetJson
         }
       } else {
-        val newOffsetJson = CosmosPartitionPlanner.createInitialOffset(container, changeFeedConfig, partitioningConfig, None)
+        val newOffsetJson = CosmosPartitionPlanner.createInitialOffset(container, containerConfig, changeFeedConfig, partitioningConfig, None)
         log.logDebug(s"No offset file location provided for batchId: $batchId " +
           s"-> offset retrieved from service: '$newOffsetJson'.")
 

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/spark/ChangeFeedBatch.scala

@@ -15,7 +15,6 @@ import org.apache.spark.sql.connector.read.ScanBuilder
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
 
 import java.util
-import java.util.UUID
 
 // scalastyle:off underscore.import
 import com.azure.cosmos.spark.CosmosTableSchemaInferrer._