fix

Author: annie-mac

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

@@ -5,6 +5,7 @@ package com.azure.cosmos.changeFeedMetrics
 
 import com.azure.cosmos.implementation.guava25.collect.EvictingQueue
 import com.azure.cosmos.spark.NormalizedRange
+import org.slf4j.{Logger, LoggerFactory}
 
 // scalastyle:off underscore.import
 import scala.collection.JavaConverters._
@@ -34,6 +35,8 @@ private[cosmos] class ChangeFeedMetricsTracker(
     private val maxHistory: Int = ChangeFeedMetricsTracker.DefaultMaxHistory,
     private val decayFactor: Double = ChangeFeedMetricsTracker.DefaultDecayFactor
 ) {
+  @transient private lazy val log : Logger = LoggerFactory.getLogger(ChangeFeedMetricsTracker.getClass)
+
   private val changeFeedChangesPerLsnHistory = EvictingQueue.create[Double](maxHistory)
   private var currentChangesPerLsnOpt: Option[Double] = None
 
@@ -83,6 +86,8 @@ private[cosmos] class ChangeFeedMetricsTracker(
    * @return Current weighted LSN gap
    */
   def getWeightedAvgChangesPerLsn: Option[Double] = {
+    // TODO: remove
+    log.info(s"getWeightedAvgChangesPerLsn for feedRangeIndex $partitionIndex, $feedRange $feedRange value $currentChangesPerLsnOpt")
     this.currentChangesPerLsnOpt
   }
 }

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

@@ -65,7 +65,7 @@ private case class ChangeFeedPartitionReader
   }
 
   private val containerTargetConfig = CosmosContainerConfig.parseCosmosContainerConfig(config)
-  log.logInfo(s"Reading from feed range ${partition.feedRange} of " +
+  log.logInfo(s"Reading from feed range ${partition.feedRange}, endLsn ${partition.endLsn} of " +
     s"container ${containerTargetConfig.database}.${containerTargetConfig.container}")
   private val readConfig = CosmosReadConfig.parseCosmosReadConfig(config)
   private val clientCacheItem = CosmosClientCache(

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

@@ -618,7 +618,7 @@ private object CosmosPartitionPlanner extends BasicLoggingTrait {
             } else {
               // the weight of this feedRange compared to other feedRanges
 
-              val feedRangeWeightFactor = this.getWeightedLsnGap(metadata, isChangeFeed, partitionMetricsMap) / totalWeightedLsnGap.get
+              val feedRangeWeightFactor = this.getWeightedLsnGap(metadata, isChangeFeed, partitionMetricsMap).toDouble / totalWeightedLsnGap.get
 
              val allowedRate =
                (feedRangeWeightFactor * maxRowsLimit.maxRows() / this.getAvgChangesPerLsn(metadata, isChangeFeed, partitionMetricsMap))