fix compiling

Author: annie-mac

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

@@ -67,7 +67,7 @@ private class ChangeFeedMicroBatchStream
   private var latestOffsetSnapshot: Option[ChangeFeedOffset] = None
 
   private val partitionIndex = new AtomicLong(0)
-  private val partitionIndexMap = Maps.synchronizedBiMap(new HashBiMap[NormalizedRange, Long]())
+  private val partitionIndexMap = Maps.synchronizedBiMap(HashBiMap.create[NormalizedRange, Long]())
   private val partitionMetricsMap = new ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]()
 
   session.sparkContext.addSparkListener(new ChangeFeedMetricsListener(partitionIndexMap, partitionMetricsMap))

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

@@ -80,7 +80,7 @@ object SparkInternalsBridge extends BasicLoggingTrait {
     }
   }
 
-  private def getInternalCustomTaskMetricsAsSQLMetricInternal(
+  private def ConsistencygetInternalCustomTaskMetricsAsSQLMetricInternal(
                                                                knownCosmosMetricNames: Set[String],
                                                                taskMetrics: TaskMetrics): Map[String, SQLMetric] = {
     getAccumulators(taskMetrics) match {

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

@@ -6,6 +6,7 @@ import com.azure.cosmos.implementation.guava25.base.MoreObjects.firstNonNull
 import com.azure.cosmos.implementation.guava25.base.Strings.emptyToNull
 import com.azure.cosmos.spark.diagnostics.BasicLoggingTrait
 import org.apache.spark.TaskContext
+import org.apache.spark.executor.TaskMetrics
 import org.apache.spark.sql.execution.metric.SQLMetric
 import org.apache.spark.util.AccumulatorV2
 
@@ -40,20 +41,23 @@ object SparkInternalsBridge extends BasicLoggingTrait {
   private final lazy val reflectionAccessAllowed = new AtomicBoolean(getSparkReflectionAccessAllowed)
 
   def getInternalCustomTaskMetricsAsSQLMetric(knownCosmosMetricNames: Set[String]) : Map[String, SQLMetric] = {
+    Option.apply(TaskContext.get()) match {
+      case Some(taskCtx) => getInternalCustomTaskMetricsAsSQLMetric(knownCosmosMetricNames, taskCtx.taskMetrics())
+      case None => Map.empty[String, SQLMetric]
+    }
+  }
+
+  def getInternalCustomTaskMetricsAsSQLMetric(knownCosmosMetricNames: Set[String], taskMetrics: TaskMetrics) : Map[String, SQLMetric] = {
 
     if (!reflectionAccessAllowed.get) {
       Map.empty[String, SQLMetric]
     } else {
-      Option.apply(TaskContext.get()) match {
-        case Some(taskCtx) => getInternalCustomTaskMetricsAsSQLMetricInternal(knownCosmosMetricNames, taskCtx)
-        case None => Map.empty[String, SQLMetric]
-      }
+      getInternalCustomTaskMetricsAsSQLMetricInternal(knownCosmosMetricNames, taskMetrics)
     }
   }
 
-  private def getAccumulators(taskCtx: TaskContext): Option[Seq[AccumulatorV2[_, _]]] = {
+  private def getAccumulators(taskMetrics: TaskMetrics): Option[Seq[AccumulatorV2[_, _]]] = {
     try {
-      val taskMetrics: Object = taskCtx.taskMetrics()
       val method = Option(accumulatorsMethod.get) match {
         case Some(existing) => existing
         case None =>
@@ -78,8 +82,8 @@ object SparkInternalsBridge extends BasicLoggingTrait {
 
   private def getInternalCustomTaskMetricsAsSQLMetricInternal(
                                                                knownCosmosMetricNames: Set[String],
-                                                               taskCtx: TaskContext): Map[String, SQLMetric] = {
-    getAccumulators(taskCtx) match {
+                                                               taskMetrics: TaskMetrics): Map[String, SQLMetric] = {
+    getAccumulators(taskMetrics) match {
       case Some(accumulators) => accumulators
         .filter(accumulable => accumulable.isInstanceOf[SQLMetric]
           && accumulable.name.isDefined

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/changeFeedMetrics/ChangeFeedMetricsTracker.scala

@@ -35,6 +35,7 @@ private[cosmos] class ChangeFeedMetricsTracker(
     private val decayFactor: Double = ChangeFeedMetricsTracker.DefaultDecayFactor
 ) {
   private val changeFeedChangesPerLsnHistory = EvictingQueue.create[Double](maxHistory)
+  private var currentChangesPerLsnOpt: Option[Double] = None
 
   /**
    * Track the normalized change feed changes per lsn
@@ -56,7 +57,7 @@ private[cosmos] class ChangeFeedMetricsTracker(
    * i = index of measurement (0 being oldest)
    * @return Weighted average of LSN gaps
    */
-  private def calculateWeightedChangesPerLsn(): Option[Double] = {
+  private def calculateWeightedChangesPerLsn(): Unit = {
     synchronized {
       if (changeFeedChangesPerLsnHistory.isEmpty) {
         None
@@ -71,8 +72,8 @@ private[cosmos] class ChangeFeedMetricsTracker(
           weightedSum += gaps(i) * weight
           weightSum += weight
         }
-        
-        Some(weightedSum / weightSum)
+
+        currentChangesPerLsnOpt = Some(weightedSum / weightSum)
       }
     }
   }
@@ -82,8 +83,6 @@ private[cosmos] class ChangeFeedMetricsTracker(
    * @return Current weighted LSN gap
    */
   def getWeightedAvgChangesPerLsn: Option[Double] = {
-    synchronized {
-      calculateWeightedChangesPerLsn()
-    }
+    this.currentChangesPerLsnOpt
   }
 }

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

@@ -132,8 +132,9 @@ private[cosmos] case class PartitionMetadata
   }
 
   def getAvgChangesPerLsn: Double = {
-    val effectiveFirstLsn = if (this.firstLsn.isEmpty) 0 else this.firstLsn.get
-    if (this.documentCount == 0 || (this.getAndValidateLatestLsn - effectiveFirstLsn) <= 0) {
+    if (this.firstLsn.isEmpty) {
+      math.max(1d, this.documentCount.toDouble / this.getAndValidateLatestLsn)
+    } else if (this.documentCount == 0 || (this.getAndValidateLatestLsn - this.firstLsn.get) <= 0) {
       1d
     } else {
       this.documentCount.toDouble / (this.getAndValidateLatestLsn- this.firstLsn.get)

sdk/cosmos/azure-cosmos-spark_3_2-12/src/test/scala/com/azure/cosmos/spark/SampleStructuredStreamingE2EMain.scala

@@ -14,8 +14,8 @@ object SampleStructuredStreamingE2EMain {
   def main(args: Array[String]) {
     val cosmosEndpoint = TestConfigurations.HOST
     val cosmosMasterKey = TestConfigurations.MASTER_KEY
-    val cosmosDatabase = "SampleDatabase"
-    val cosmosContainer = "GreenTaxiRecords"
+    val cosmosDatabase = "TestDatabase"
+    val cosmosContainer = "TestContainerMulti"
 
     //    val client = new CosmosClientBuilder()
     //      .endpoint(cosmosEndpoint)
@@ -53,15 +53,7 @@ object SampleStructuredStreamingE2EMain {
       "spark.cosmos.read.inferSchema.enabled" -> "false",
       "spark.cosmos.changeFeed.startFrom" -> "Beginning",
       "spark.cosmos.changeFeed.mode" -> "Incremental",
-      "spark.cosmos.changeFeed.itemCountPerTriggerHint" -> "10000",
-      "spark.cosmos.preferredRegionsList" -> "[West Europe, North Europe]",
-      "spark.cosmos.throughputControl.enabled" -> "true",
-      "spark.cosmos.throughputControl.name" -> "CopyReadFromSource",
-      "spark.cosmos.throughputControl.targetThroughputThreshold" -> "0.95",
-      "spark.cosmos.throughputControl.globalControl.database" -> "SampleDatabase",
-      "spark.cosmos.throughputControl.globalControl.container" -> "ThroughputControl",
-      "spark.cosmos.clientTelemetry.enabled" -> "true",
-      "spark.cosmos.read.partitioning.strategy" -> "Restrictive",
+      "spark.cosmos.changeFeed.itemCountPerTriggerHint" -> "1",
       "spark.cosmos.diagnostics" -> "feed"
     )
     val changeFeed_df = spark.readStream.format("cosmos.oltp.changeFeed").options(changeFeedCfg).load()