Azure · FabianMeiswinkel · Jan 15, 2026 · Jan 15, 2026 · Jan 15, 2026 · Jan 15, 2026
@@ -106,6 +106,7 @@
     "sdk/cosmos/azure-cosmos-spark_3-5/**",
     "sdk/cosmos/azure-cosmos-spark_3-5_2-12/**",
     "sdk/cosmos/azure-cosmos-spark_3-5_2-13/**",
+    "sdk/cosmos/azure-cosmos-spark_4-0_2-13/**",
     "sdk/cosmos/azure-cosmos-spark-account-data-resolver-sample/**",
     "sdk/cosmos/fabric-cosmos-spark-auth_3/**",
     "sdk/cosmos/azure-cosmos-encryption/**",

@@ -79,6 +79,7 @@ known_content_issues:
   - ['sdk/cosmos/azure-cosmos-spark_3-4_2-12/README.md', '#3113']
   - ['sdk/cosmos/azure-cosmos-spark_3-5_2-12/README.md', '#3113']
   - ['sdk/cosmos/azure-cosmos-spark_3-5_2-13/README.md', '#3113']
+  - ['sdk/cosmos/azure-cosmos-spark_4-0_2-13/README.md', '#3113']
   - ['sdk/cosmos/azure-cosmos-spark-account-data-resolver-sample/README.md', '#3113']
   - ['sdk/cosmos/fabric-cosmos-spark-auth_3/README.md', '#3113']
   - ['sdk/cosmos/azure-cosmos-spark_3_2-12/dev/README.md', '#3113']

@@ -48,7 +48,7 @@ extends:
               displayName: 'Build all libraries that support Java $(JavaBuildVersion)'
               inputs:
                 mavenPomFile: pom.xml
-                options: '$(DefaultOptions) -T 2C -DskipTests -Dgpg.skip -Dmaven.javadoc.skip=true -Dcodesnippet.skip=true -Dcheckstyle.skip=true -Dspotbugs.skip=true -Djacoco.skip=true -Drevapi.skip=true -Dshade.skip=true -Dspotless.skip=true -pl !com.azure.cosmos.spark:azure-cosmos-spark_3-3_2-12,!com.azure.cosmos.spark:azure-cosmos-spark_3-4_2-12,!com.azure.cosmos.spark:azure-cosmos-spark_3-5_2-12,!com.azure.cosmos.spark:azure-cosmos-spark-account-data-resolver-sample,!com.azure.cosmos.kafka:azure-cosmos-kafka-connect,!com.microsoft.azure:azure-batch'
+                options: '$(DefaultOptions) -T 2C -DskipTests -Dgpg.skip -Dmaven.javadoc.skip=true -Dcodesnippet.skip=true -Dcheckstyle.skip=true -Dspotbugs.skip=true -Djacoco.skip=true -Drevapi.skip=true -Dshade.skip=true -Dspotless.skip=true -pl !com.azure.cosmos.spark:azure-cosmos-spark_3-3_2-12,!com.azure.cosmos.spark:azure-cosmos-spark_3-4_2-12,!com.azure.cosmos.spark:azure-cosmos-spark_3-5_2-12,!com.azure.cosmos.spark:azure-cosmos-spark_3-5_2-13,!com.azure.cosmos.spark:azure-cosmos-spark_4-0_2-13,!com.azure.cosmos.spark:azure-cosmos-spark-account-data-resolver-sample,!com.azure.cosmos.kafka:azure-cosmos-kafka-connect,!com.microsoft.azure:azure-batch'
                 mavenOptions: '$(MemoryOptions) $(LoggingOptions)'
                 javaHomeOption: 'JDKVersion'
                 jdkVersionOption: $(JavaBuildVersion)

@@ -92,6 +92,20 @@
         "JavaTestVersion": "1.17",
         "AdditionalArgs": "-DACCOUNT_HOST=https://localhost:8081/ -Dhadoop.home.dir=D:/Hadoop -DCOSMOS.AZURE_COSMOS_DISABLE_NON_STREAMING_ORDER_BY=true"
       },
+      "Spark 4.0, Scala 2.13 Integration Tests targeting Cosmos Emulator - Java 17'": {
+        "ProfileFlag": "-Dspark-e2e_4-0_2-13",
+        "PROTOCOLS": "[\"Tcp\"]",
+        "DESIRED_CONSISTENCIES": "[\"Session\"]",
+        "JavaTestVersion": "1.17",
+        "AdditionalArgs": "-DACCOUNT_HOST=https://localhost:8081/ -Dhadoop.home.dir=D:/Hadoop -DCOSMOS.AZURE_COSMOS_DISABLE_NON_STREAMING_ORDER_BY=true"
+      },
+      "Spark 4.0, Scala 2.13 Integration Tests targeting Cosmos Emulator - Java 21'": {
+        "ProfileFlag": "-Dspark-e2e_4-0_2-13",
+        "PROTOCOLS": "[\"Tcp\"]",
+        "DESIRED_CONSISTENCIES": "[\"Session\"]",
+        "JavaTestVersion": "1.21",
+        "AdditionalArgs": "-DACCOUNT_HOST=https://localhost:8081/ -Dhadoop.home.dir=D:/Hadoop -DCOSMOS.AZURE_COSMOS_DISABLE_NON_STREAMING_ORDER_BY=true"
+      },
       "Kafka Integration Tests targeting Cosmos Emulator - Java 11": {
         "ProfileFlag": "-Pkafka-emulator",
         "PROTOCOLS": "[\"Tcp\"]",

@@ -125,7 +125,34 @@ extends:
                   keytool -keystore cacerts -delete -noprompt -alias CosmosDbEmulatorCert -storepass changeit
                   keytool -keystore cacerts -importcert -noprompt -trustcacerts -alias CosmosDbEmulatorCert -file CosmosDbEmulatorCert.cer -storepass changeit
                 displayName: 'Create Java TrustStore'
-
+              PostSteps:
+                - powershell: |
+                    Write-Host "===== SCALATEST REPORTS (TXT) ====="
+                    Get-ChildItem -Path "$(System.DefaultWorkingDirectory)" -Recurse -Filter "*.txt" | Where-Object { $_.FullName -like "*surefire-reports*" } | ForEach-Object {
+                      Write-Host "File: $($_.FullName)"
+                      Get-Content $_.FullName
+                    }
+
+                    Write-Host "===== SCALATEST REPORTS (OUT) ====="
+                    Get-ChildItem -Path "$(System.DefaultWorkingDirectory)" -Recurse -Filter "*.out" | Where-Object { $_.FullName -like "*surefire-reports*" } | ForEach-Object {
+                      Write-Host "File: $($_.FullName)"
+                      Get-Content $_.FullName
+                    }
+
+                    Write-Host "===== SCALATEST REPORTS (ERR) ====="
+                    Get-ChildItem -Path "$(System.DefaultWorkingDirectory)" -Recurse -Filter "*.err" | Where-Object { $_.FullName -like "*surefire-reports*" } | ForEach-Object {
+                      Write-Host "File: $($_.FullName)"
+                      Get-Content $_.FullName
+                    }
+
+                    Write-Host "===== SCALATEST XML (RAW) ====="
+                    Get-ChildItem -Path "$(System.DefaultWorkingDirectory)" -Recurse -Filter "*.xml" | Where-Object { $_.FullName -like "*surefire-reports*" } | ForEach-Object {
+                      Write-Host "File: $($_.FullName)"
+                      Get-Content $_.FullName
+                    }
+                  displayName: "Print ScalaTest report files"
+                  errorActionPreference: continue
+                  condition: always()
       - ${{ if eq(variables['System.TeamProject'], 'internal') }}:
         - ${{ each mode in parameters.VnextEmulatorModes }}:
           - stage:

@@ -20,6 +20,7 @@ com.fasterxml.jackson.datatype:jackson-datatype-jsr310;2.18.4
 com.fasterxml.jackson.module:jackson-module-afterburner;2.18.4
 com.fasterxml.jackson.module:jackson-module-blackbird;2.18.4
 com.fasterxml.jackson.module:jackson-module-scala_2.12;2.18.4
+com.fasterxml.jackson.module:jackson-module-scala_2.13;2.18.4
 com.github.spotbugs:spotbugs;4.8.3
 com.github.spotbugs:spotbugs-annotations;4.8.3
 com.github.spotbugs:spotbugs-maven-plugin;4.8.3.1
@@ -239,15 +240,14 @@ cosmos_io.dropwizard.metrics:metrics-graphite;4.1.0
 cosmos_io.dropwizard.metrics:metrics-jvm;4.1.0
 cosmos_org.mpierce.metrics.reservoir:hdrhistogram-metrics-reservoir;1.1.0
 cosmos_org.hdrhistogram:HdrHistogram;2.1.12
-cosmos_com.fasterxml.jackson.core:jackson-databind;2.15.2
-cosmos_com.fasterxml.jackson.module:jackson-module-scala_2.12;2.15.2
 cosmos_com.microsoft.azure.synapse:synapseutils_2.12;1.5.4
 
 ## Cosmos Spark connector under sdk\cosmos\azure-cosmos-spark_3-<version>_2-12\pom.xml
 # Cosmos Spark connector runtime dependencies - provided by Spark runtime/host
 cosmos-spark_3-3_org.apache.spark:spark-sql_2.12;3.3.0
 cosmos-spark_3-4_org.apache.spark:spark-sql_2.12;3.4.0
 cosmos-spark_3-5_org.apache.spark:spark-sql_2.12;3.5.0
+cosmos-spark_4-0_org.apache.spark:spark-sql_2.13;4.0.0
 cosmos-spark_3-3_org.apache.spark:spark-hive_2.12;3.3.0
 cosmos-spark_3-4_org.apache.spark:spark-hive_2.12;3.4.0
 cosmos-spark_3-5_org.apache.spark:spark-hive_2.12;3.5.0

diff --git a/eng/versioning/version_client.txt b/eng/versioning/version_client.txt
@@ -115,7 +115,8 @@ com.azure:azure-cosmos-test;1.0.0-beta.16;1.0.0-beta.17
 com.azure.cosmos.spark:azure-cosmos-spark_3-3_2-12;4.42.0;4.43.0-beta.1
 com.azure.cosmos.spark:azure-cosmos-spark_3-4_2-12;4.42.0;4.43.0-beta.1
 com.azure.cosmos.spark:azure-cosmos-spark_3-5_2-12;4.42.0;4.43.0-beta.1
-com.azure.cosmos.spark:azure-cosmos-spark_3-5_2-13;4.42.0;4.43.0-beta.1
+com.azure.cosmos.spark:azure-cosmos-spark_3-5_2-13;4.43.0-beta.1;4.43.0-beta.1
+com.azure.cosmos.spark:azure-cosmos-spark_4-0_2-13;4.43.0-beta.1;4.43.0-beta.1
 com.azure.cosmos.spark:fabric-cosmos-spark-auth_3;1.1.0;1.2.0-beta.1
 com.azure:azure-cosmos-tests;1.0.0-beta.1;1.0.0-beta.1
 com.azure:azure-data-appconfiguration;1.9.0;1.10.0-beta.1

@@ -75,12 +75,12 @@
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-databind</artifactId>
-      <version>2.15.2</version> <!-- {x-version-update;cosmos_com.fasterxml.jackson.core:jackson-databind;external_dependency} -->
+      <version>2.18.4</version> <!-- {x-version-update;com.fasterxml.jackson.core:jackson-databind;external_dependency} -->
     </dependency>
     <dependency>
       <groupId>com.fasterxml.jackson.module</groupId>
       <artifactId>jackson-module-scala_2.12</artifactId>
-      <version>2.15.2</version> <!-- {x-version-update;cosmos_com.fasterxml.jackson.module:jackson-module-scala_2.12;external_dependency} -->
+      <version>2.18.4</version> <!-- {x-version-update;com.fasterxml.jackson.module:jackson-module-scala_2.12;external_dependency} -->
     </dependency>
     <dependency>
       <groupId>org.scala-lang</groupId>
@@ -289,8 +289,6 @@
                 <include>com.fasterxml.jackson.datatype:jackson-datatype-jsr310:[2.18.4]</include> <!-- {x-include-update;com.fasterxml.jackson.datatype:jackson-datatype-jsr310;external_dependency} -->
                 <include>com.fasterxml.jackson.core:jackson-databind:[2.18.4]</include> <!-- {x-include-update;com.fasterxml.jackson.core:jackson-databind;external_dependency} -->
                 <include>com.fasterxml.jackson.module:jackson-module-scala_2.12:[2.18.4]</include> <!-- {x-include-update;com.fasterxml.jackson.module:jackson-module-scala_2.12;external_dependency} -->
-                <include>com.fasterxml.jackson.core:jackson-databind:[2.15.2]</include> <!-- {x-include-update;cosmos_com.fasterxml.jackson.core:jackson-databind;external_dependency} -->
-                <include>com.fasterxml.jackson.module:jackson-module-scala_2.12:[2.15.2]</include> <!-- {x-include-update;cosmos_com.fasterxml.jackson.module:jackson-module-scala_2.12;external_dependency} -->
                 <include>com.globalmentor:hadoop-bare-naked-local-fs:[0.1.0]</include> <!-- {x-include-update;cosmos_com.globalmentor:hadoop-bare-naked-local-fs;external_dependency} -->
                 <include>com.azure.cosmos.spark:azure-cosmos-spark_3-5_2-12:[4.43.0-beta.1]</include> <!-- {x-include-update;com.azure.cosmos.spark:azure-cosmos-spark_3-5_2-12;current} -->
               </includes>

@@ -148,6 +148,12 @@
             <artifactId>scalatest-maven-plugin</artifactId>
             <version>2.1.0</version> <!-- {x-version-update;cosmos_org.scalatest:scalatest-maven-plugin;external_dependency} -->
             <configuration>
+              <config>stdOut=true,verbose=true,stdErr=true</config>
+              <noScalaTestIgnore>true</noScalaTestIgnore>
+              <stdout>FDEF</stdout>
+              <stderr>FDEF</stderr>
+              <forkMode>once</forkMode>
+              <logForkedProcessCommand>true</logForkedProcessCommand>
               <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
               <junitxml>.</junitxml>
               <filereports>SparkTestSuite.txt</filereports>

@@ -1,5 +1,7 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
+// scalastyle:off magic.number
+// scalastyle:off multiple.string.literals
 
 package com.azure.cosmos.spark
 

@@ -148,6 +148,14 @@
             <artifactId>scalatest-maven-plugin</artifactId>
             <version>2.1.0</version> <!-- {x-version-update;cosmos_org.scalatest:scalatest-maven-plugin;external_dependency} -->
             <configuration>
+              <config>stdOut=true,verbose=true,stdErr=true</config>
+              <noScalaTestIgnore>true</noScalaTestIgnore>
+              <stdout>FDEF</stdout>
+              <stderr>FDEF</stderr>
+              <config>stdOut=true,verbose=true,stdErr=true</config>
+              <noScalaTestIgnore>true</noScalaTestIgnore>
+              <forkMode>once</forkMode>
+              <logForkedProcessCommand>true</logForkedProcessCommand>
               <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
               <junitxml>.</junitxml>
               <filereports>SparkTestSuite.txt</filereports>

@@ -0,0 +1,140 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+// scalastyle:off magic.number
+// scalastyle:off multiple.string.literals
+
+package com.azure.cosmos.spark
+
+import com.azure.cosmos.changeFeedMetrics.{ChangeFeedMetricsListener, ChangeFeedMetricsTracker}
+import com.azure.cosmos.implementation.guava25.collect.{HashBiMap, Maps}
+import org.apache.spark.Success
+import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
+import org.apache.spark.scheduler.{SparkListenerTaskEnd, TaskInfo}
+import org.apache.spark.sql.execution.metric.SQLMetric
+import org.mockito.ArgumentMatchers
+import org.mockito.Mockito.{mock, when}
+
+import java.lang.reflect.Field
+import java.util.concurrent.ConcurrentHashMap
+
+class ChangeFeedMetricsListenerSpec extends UnitSpec {
+  "ChangeFeedMetricsListener" should "be able to capture changeFeed performance metrics" in {
+   val taskEnd = SparkListenerTaskEnd(
+    stageId = 1,
+    stageAttemptId = 0,
+    taskType = "ResultTask",
+    reason = Success,
+    taskInfo = mock(classOf[TaskInfo]),
+    taskExecutorMetrics = mock(classOf[ExecutorMetrics]),
+    taskMetrics = mock(classOf[TaskMetrics])
+   )
+
+   val metrics = Map[String, SQLMetric](
+    CosmosConstants.MetricNames.ChangeFeedPartitionIndex -> new SQLMetric("index", 1),
+    CosmosConstants.MetricNames.ChangeFeedLsnRange -> new SQLMetric("lsn", 100),
+    CosmosConstants.MetricNames.ChangeFeedItemsCnt -> new SQLMetric("items", 100)
+   )
+
+   // create sparkInternalsBridge mock
+   val sparkInternalsBridge = mock(classOf[SparkInternalsBridge])
+   when(sparkInternalsBridge.getInternalCustomTaskMetricsAsSQLMetric(
+    ArgumentMatchers.any[Set[String]],
+    ArgumentMatchers.any[TaskMetrics]
+   )).thenReturn(metrics)
+
+   val partitionIndexMap = Maps.synchronizedBiMap(HashBiMap.create[NormalizedRange, Long]())
+   partitionIndexMap.put(NormalizedRange("0", "FF"), 1)
+
+   val partitionMetricsMap = new ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]()
+   val changeFeedMetricsListener = new ChangeFeedMetricsListener(partitionIndexMap, partitionMetricsMap)
+
+   // set the internal sparkInternalsBridgeField
+   val sparkInternalsBridgeField: Field = classOf[ChangeFeedMetricsListener].getDeclaredField("sparkInternalsBridge")
+   sparkInternalsBridgeField.setAccessible(true)
+   sparkInternalsBridgeField.set(changeFeedMetricsListener, sparkInternalsBridge)
+
+   //  verify that metrics will be properly tracked
+   changeFeedMetricsListener.onTaskEnd(taskEnd)
+   partitionMetricsMap.size() shouldBe 1
+   partitionMetricsMap.containsKey(NormalizedRange("0", "FF")) shouldBe true
+   partitionMetricsMap.get(NormalizedRange("0", "FF")).getWeightedChangeFeedItemsPerLsn.get shouldBe 1
+  }
+
+ it should "ignore metrics for unknown partition index" in {
+  val taskEnd = SparkListenerTaskEnd(
+   stageId = 1,
+   stageAttemptId = 0,
+   taskType = "ResultTask",
+   reason = Success,
+   taskInfo = mock(classOf[TaskInfo]),
+   taskExecutorMetrics = mock(classOf[ExecutorMetrics]),
+   taskMetrics = mock(classOf[TaskMetrics])
+  )
+
+  val metrics = Map[String, SQLMetric](
+   CosmosConstants.MetricNames.ChangeFeedPartitionIndex -> new SQLMetric("index", 10),
+   CosmosConstants.MetricNames.ChangeFeedLsnRange -> new SQLMetric("lsn", 100),
+   CosmosConstants.MetricNames.ChangeFeedItemsCnt -> new SQLMetric("items", 100)
+  )
+
+  // create sparkInternalsBridge mock
+  val sparkInternalsBridge = mock(classOf[SparkInternalsBridge])
+  when(sparkInternalsBridge.getInternalCustomTaskMetricsAsSQLMetric(
+   ArgumentMatchers.any[Set[String]],
+   ArgumentMatchers.any[TaskMetrics]
+  )).thenReturn(metrics)
+
+  val partitionIndexMap = Maps.synchronizedBiMap(HashBiMap.create[NormalizedRange, Long]())
+  partitionIndexMap.put(NormalizedRange("0", "FF"), 1)
+
+  val partitionMetricsMap = new ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]()
+  val changeFeedMetricsListener = new ChangeFeedMetricsListener(partitionIndexMap, partitionMetricsMap)
+
+  // set the internal sparkInternalsBridgeField
+  val sparkInternalsBridgeField: Field = classOf[ChangeFeedMetricsListener].getDeclaredField("sparkInternalsBridge")
+  sparkInternalsBridgeField.setAccessible(true)
+  sparkInternalsBridgeField.set(changeFeedMetricsListener, sparkInternalsBridge)
+
+  // because partition index 10 does not exist in the partitionIndexMap, it will be ignored
+  changeFeedMetricsListener.onTaskEnd(taskEnd)
+  partitionMetricsMap shouldBe empty
+ }
+
+ it should "ignore unrelated metrics" in {
+  val taskEnd = SparkListenerTaskEnd(
+   stageId = 1,
+   stageAttemptId = 0,
+   taskType = "ResultTask",
+   reason = Success,
+   taskInfo = mock(classOf[TaskInfo]),
+   taskExecutorMetrics = mock(classOf[ExecutorMetrics]),
+   taskMetrics = mock(classOf[TaskMetrics])
+  )
+
+  val metrics = Map[String, SQLMetric](
+   "unknownMetrics" -> new SQLMetric("index", 10)
+  )
+
+  // create sparkInternalsBridge mock
+  val sparkInternalsBridge = mock(classOf[SparkInternalsBridge])
+  when(sparkInternalsBridge.getInternalCustomTaskMetricsAsSQLMetric(
+   ArgumentMatchers.any[Set[String]],
+   ArgumentMatchers.any[TaskMetrics]
+  )).thenReturn(metrics)
+
+  val partitionIndexMap = Maps.synchronizedBiMap(HashBiMap.create[NormalizedRange, Long]())
+  partitionIndexMap.put(NormalizedRange("0", "FF"), 1)
+
+  val partitionMetricsMap = new ConcurrentHashMap[NormalizedRange, ChangeFeedMetricsTracker]()
+  val changeFeedMetricsListener = new ChangeFeedMetricsListener(partitionIndexMap, partitionMetricsMap)
+
+  // set the internal sparkInternalsBridgeField
+  val sparkInternalsBridgeField: Field = classOf[ChangeFeedMetricsListener].getDeclaredField("sparkInternalsBridge")
+  sparkInternalsBridgeField.setAccessible(true)
+  sparkInternalsBridgeField.set(changeFeedMetricsListener, sparkInternalsBridge)
+
+  // because partition index 10 does not exist in the partitionIndexMap, it will be ignored
+  changeFeedMetricsListener.onTaskEnd(taskEnd)
+  partitionMetricsMap shouldBe empty
+ }
+}