refactor

Author: annie-mac

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

@@ -10,7 +10,7 @@
 * Added log4j-core to the list of shaded packages to avoid conflicts when customers use log4j in a different version. - See [PR 45924](https://github.com/Azure/azure-sdk-for-java/pull/46451)
 
 #### Other Changes
-* Added improvement for end lsn calculation in `CosmosPartitionPlanner`. - See [PR 46320](https://github.com/Azure/azure-sdk-for-java/pull/46320)
+* Added change feed performance monitoring which is used to improve end lsn calculation in `CosmosPartitionPlanner`. - See [PR 46320](https://github.com/Azure/azure-sdk-for-java/pull/46320)
 
 ### 4.38.0 (2025-07-31)
 

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

@@ -69,11 +69,11 @@ private class ChangeFeedMicroBatchStream
   private val partitionMetricsMap = new ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]()
 
   // Register metrics listener
-  if (CosmosConstants.ChangeFeedMetricsListenerConfig.metricsListenerEnabled) {
-    log.logInfo("Register ChangeFeedMetricsListener")
+  if (changeFeedConfig.performanceMonitoringEnabled) {
+    log.logInfo("ChangeFeed performance monitoring is enabled, registering ChangeFeedMetricsListener")
     session.sparkContext.addSparkListener(new ChangeFeedMetricsListener(partitionIndexMap, partitionMetricsMap))
   } else {
-    log.logInfo("ChangeFeedMetricsListener is disabled")
+    log.logInfo("ChangeFeed performance monitoring is disabled")
   }
 
   override def latestOffset(): Offset = {

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

@@ -10,7 +10,7 @@
 * Added log4j-core to the list of shaded packages to avoid conflicts when customers use log4j in a different version. - See [PR 45924](https://github.com/Azure/azure-sdk-for-java/pull/46451)
 
 #### Other Changes
-* Added improvement for end lsn calculation in `CosmosPartitionPlanner`. - See [PR 46320](https://github.com/Azure/azure-sdk-for-java/pull/46320)
+* Added change feed performance monitoring which is used to improve end lsn calculation in `CosmosPartitionPlanner`. - See [PR 46320](https://github.com/Azure/azure-sdk-for-java/pull/46320)
 
 ### 4.38.0 (2025-07-31)
 

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

@@ -70,11 +70,11 @@ private class ChangeFeedMicroBatchStream
   private val partitionIndexMap = Maps.synchronizedBiMap(HashBiMap.create[NormalizedRange, Long]())
   private val partitionMetricsMap = new ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]()
 
-  if (CosmosConstants.ChangeFeedMetricsListenerConfig.metricsListenerEnabled) {
-    log.logInfo("Register ChangeFeedMetricsListener")
+  if (changeFeedConfig.performanceMonitoringEnabled) {
+    log.logInfo("ChangeFeed performance monitoring is enabled, registering ChangeFeedMetricsListener")
     session.sparkContext.addSparkListener(new ChangeFeedMetricsListener(partitionIndexMap, partitionMetricsMap))
   } else {
-    log.logInfo("ChangeFeedMetricsListener is disabled")
+    log.logInfo("ChangeFeed performance monitoring is disabled")
   }
 
   override def latestOffset(): Offset = {

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

@@ -10,7 +10,7 @@
 * Added log4j-core to the list of shaded packages to avoid conflicts when customers use log4j in a different version. - See [PR 45924](https://github.com/Azure/azure-sdk-for-java/pull/46451)
 
 #### Other Changes
-* Added improvement for end lsn calculation in `CosmosPartitionPlanner`. - See [PR 46320](https://github.com/Azure/azure-sdk-for-java/pull/46320)
+* Added change feed performance monitoring which is used to improve end lsn calculation in `CosmosPartitionPlanner`. - See [PR 46320](https://github.com/Azure/azure-sdk-for-java/pull/46320)
 
 ### 4.38.0 (2025-07-31)
 

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

@@ -71,11 +71,11 @@ private class ChangeFeedMicroBatchStream
   private val partitionIndexMap = Maps.synchronizedBiMap(HashBiMap.create[NormalizedRange, Long]())
   private val partitionMetricsMap = new ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]()
 
-  if (CosmosConstants.ChangeFeedMetricsListenerConfig.metricsListenerEnabled) {
-    log.logInfo("Register ChangeFeedMetricsListener")
+  if (changeFeedConfig.performanceMonitoringEnabled) {
+    log.logInfo("ChangeFeed performance monitoring is enabled, registering ChangeFeedMetricsListener")
     session.sparkContext.addSparkListener(new ChangeFeedMetricsListener(partitionIndexMap, partitionMetricsMap))
   } else {
-    log.logInfo("ChangeFeedMetricsListener is disabled")
+    log.logInfo("ChangeFeed performance monitoring is disabled")
   }
 
   override def latestOffset(): Offset = {

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

@@ -0,0 +1,18 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos.changeFeedMetrics
+
+import com.azure.cosmos.spark.CosmosConstants.MetricNames
+import org.apache.spark.sql.connector.metric.CustomSumMetric
+
+/***
+ * This metric is used to track the partition total items count within a change feed micro-batch
+ * Note: not all the items will be returned back to spark
+ */
+private[cosmos] class ChangeFeedItemsCntMetric extends CustomSumMetric {
+
+ override def name(): String = MetricNames.ChangeFeedItemsCnt
+
+ override def description(): String = MetricNames.ChangeFeedItemsCnt
+}

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

@@ -20,7 +20,7 @@ private[cosmos] class ChangeFeedMetricsListener(
    val metrics = SparkInternalsBridge.getInternalCustomTaskMetricsAsSQLMetric(
     Set(
      CosmosConstants.MetricNames.ChangeFeedLsnRange,
-     CosmosConstants.MetricNames.ChangeFeedFetchedChangesCnt,
+     CosmosConstants.MetricNames.ChangeFeedItemsCnt,
      CosmosConstants.MetricNames.ChangeFeedPartitionIndex
     ),
     taskEnd.taskMetrics
@@ -31,19 +31,19 @@ private[cosmos] class ChangeFeedMetricsListener(
 
     val normalizedRange = partitionIndexMap.inverse().get(index)
     if (normalizedRange != null) {
-     partitionMetricsMap.putIfAbsent(normalizedRange, new ChangeFeedMetricsTracker(index, normalizedRange))
+     partitionMetricsMap.putIfAbsent(normalizedRange, ChangeFeedMetricsTracker(index, normalizedRange))
      val metricsTracker = partitionMetricsMap.get(normalizedRange)
-     val changesCnt = getChangesCnt(metrics)
+     val fetchedItemCnt = getFetchedItemCnt(metrics)
      val lsnRange = getLsnRange(metrics)
 
-     if (changesCnt >= 0 && lsnRange >= 0) {
+     if (fetchedItemCnt >= 0 && lsnRange >= 0) {
       metricsTracker.track(
        metrics(CosmosConstants.MetricNames.ChangeFeedLsnRange).value,
-       metrics(CosmosConstants.MetricNames.ChangeFeedFetchedChangesCnt).value
+       metrics(CosmosConstants.MetricNames.ChangeFeedItemsCnt).value
       )
      }
 
-     logInfo(s"onTaskEnd for partition index $index, changesCnt $changesCnt, lsnRange $lsnRange")
+     logInfo(s"onTaskEnd for partition index $index, fetchedItemCnt $fetchedItemCnt, lsnRange $lsnRange")
     }
    }
   } catch {
@@ -54,15 +54,15 @@ private[cosmos] class ChangeFeedMetricsListener(
   }
  }
 
- def getChangesCnt(metrics: Map[String, SQLMetric]): Long = {
-  if (metrics.contains(CosmosConstants.MetricNames.ChangeFeedFetchedChangesCnt)) {
-   metrics(CosmosConstants.MetricNames.ChangeFeedFetchedChangesCnt).value
+ private def getFetchedItemCnt(metrics: Map[String, SQLMetric]): Long = {
+  if (metrics.contains(CosmosConstants.MetricNames.ChangeFeedItemsCnt)) {
+   metrics(CosmosConstants.MetricNames.ChangeFeedItemsCnt).value
   } else {
    -1
   }
  }
 
- def getLsnRange(metrics: Map[String, SQLMetric]): Long = {
+ private def getLsnRange(metrics: Map[String, SQLMetric]): Long = {
   if (metrics.contains(CosmosConstants.MetricNames.ChangeFeedLsnRange)) {
    metrics(CosmosConstants.MetricNames.ChangeFeedLsnRange).value
   } else {

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

@@ -12,76 +12,81 @@ import scala.collection.JavaConverters._
 // scalastyle:on underscore.import
 
 private[cosmos] object ChangeFeedMetricsTracker {
-  private val DefaultMaxHistory: Int = CosmosConstants.ChangeFeedTrackerConfigs.changeFeedMetricsTrackerMaxHistory
-  private val DefaultDecayFactor: Double = CosmosConstants.ChangeFeedTrackerConfigs.changeFeedMetricsTrackerDecayFactor
+  private val MetricsHistory: Int = CosmosConstants.ChangeFeedMetricsConfigs.MetricsHistory
+  private val MetricsHistoryDecayFactor: Double = CosmosConstants.ChangeFeedMetricsConfigs.MetricsHistoryDecayFactor
 
-  def apply(partitionIndex: Long, feedRange: NormalizedRange): ChangeFeedMetricsTracker = {
-    new ChangeFeedMetricsTracker(partitionIndex, feedRange)
+  def apply(
+            partitionIndex: Long,
+            feedRange: NormalizedRange,
+            metricsHistory: Int,
+            metricsHistoryDecayFactor: Double): ChangeFeedMetricsTracker = {
+   new ChangeFeedMetricsTracker(partitionIndex, feedRange, MetricsHistory, MetricsHistoryDecayFactor)
   }
 
-  def apply(partitionIndex: Long, feedRange: NormalizedRange, decayFactor: Double): ChangeFeedMetricsTracker = {
-    new ChangeFeedMetricsTracker(partitionIndex, feedRange, DefaultMaxHistory, decayFactor)
+  def apply(partitionIndex: Long, feedRange: NormalizedRange): ChangeFeedMetricsTracker = {
+    new ChangeFeedMetricsTracker(partitionIndex, feedRange, MetricsHistory, MetricsHistoryDecayFactor)
   }
 }
 
 /**
- * Tracks metrics for Change Feed operations including partition indices and LSN gaps
+ * Tracks metrics for Change Feed operations including partition indices and LSN range
  * with exponential weighting for recent metrics
  */
 private[cosmos] class ChangeFeedMetricsTracker(
     private val partitionIndex: Long,
     private val feedRange: NormalizedRange,
-    private val maxHistory: Int = ChangeFeedMetricsTracker.DefaultMaxHistory,
-    private val decayFactor: Double = ChangeFeedMetricsTracker.DefaultDecayFactor) extends BasicLoggingTrait {
+    private val maxHistory: Int = ChangeFeedMetricsTracker.MetricsHistory,
+    private val decayFactor: Double = ChangeFeedMetricsTracker.MetricsHistoryDecayFactor) extends BasicLoggingTrait {
 
-  private val changeFeedChangesPerLsnHistory = EvictingQueue.create[Double](maxHistory)
-  private var currentChangesPerLsnOpt: Option[Double] = None
+  private val changeFeedItemsPerLsnHistory = EvictingQueue.create[Double](maxHistory)
+  private var currentChangeFeedItemsPerLsnOpt: Option[Double] = None
 
   /**
-   * Track the normalized change feed changes per lsn
+   * Track the normalized change feed items per lsn
    *
-   * @param lsnRange the lsn range of the fetched changes.
-   * @param changesFetchedCnt the total fetched changes.
+   * @param lsnRange the lsn range of the fetched items.
+   * @param fetchedItemCnt the total fetched item cnt within a micro-batch for the feed range.
    */
-  def track(lsnRange: Long, changesFetchedCnt: Long): Unit = {
-    val effectiveChangesFetchedCnt = Math.max(1, changesFetchedCnt)
+  def track(lsnRange: Long, fetchedItemCnt: Long): Unit = {
+    val effectiveChangesFetchedCnt = Math.max(1, fetchedItemCnt)
     val changesPerLsn = if (lsnRange == 0) effectiveChangesFetchedCnt.toDouble else effectiveChangesFetchedCnt.toDouble / lsnRange
     synchronized {
-      changeFeedChangesPerLsnHistory.add(changesPerLsn)
-      calculateWeightedChangesPerLsn()
+      changeFeedItemsPerLsnHistory.add(changesPerLsn)
+      calculateWeightedItemsPerLsn()
     }
   }
 
   /**
-   * Calculates weighted normalized changes per lsn where recent values have more impact
+   * Calculates weighted normalized items per lsn where recent values have more impact
    * Uses exponential weighting: weight = decayFactor^(n-i-1) where:
    * n = number of measurements
    * i = index of measurement (0 being oldest)
    * @return Weighted changes per lsn
    */
-  private def calculateWeightedChangesPerLsn(): Unit = {
-    if (!changeFeedChangesPerLsnHistory.isEmpty) {
-      val gaps = changeFeedChangesPerLsnHistory.asScala.toArray
-      val n = gaps.length
+  private def calculateWeightedItemsPerLsn(): Unit = {
+    if (!changeFeedItemsPerLsnHistory.isEmpty) {
+      val histories = changeFeedItemsPerLsnHistory.asScala.toArray
+      val n = histories.length
       var weightedSum = 0.0
       var weightSum = 0.0
 
-      for (i <- gaps.indices) {
+      for (i <- histories.indices) {
         val weight = math.pow(decayFactor, n - i - 1)
-        weightedSum += gaps(i) * weight
+        weightedSum += histories(i) * weight
         weightSum += weight
       }
 
-      currentChangesPerLsnOpt = Some(weightedSum / weightSum)
+      currentChangeFeedItemsPerLsnOpt = Some(weightedSum / weightSum)
     }
   }
 
   /**
-   * Gets current weighted changes per lsn
-   * @return weighted changes per lsn
+   * Gets current weighted items per lsn
+   * @return weighted items per lsn
    */
-  def getWeightedChangesPerLsn: Option[Double] = {
-    logDebug(s"getWeightedChangesPerLsn for feedRangeIndex [$partitionIndex], feedRange [$feedRange] value [$currentChangesPerLsnOpt]")
-    this.currentChangesPerLsnOpt
+  def getWeightedChangeFeedItemsPerLsn: Option[Double] = {
+    logDebug(s"getWeightedChangeFeedItemsPerLsn for feedRangeIndex [$partitionIndex], " +
+     s"feedRange [$feedRange] value [$currentChangeFeedItemsPerLsnOpt]")
+    this.currentChangeFeedItemsPerLsnOpt
   }
 }