From 8be5fd2ebe52f4cad34d9cb104ad60791e39c4d9 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 16 Apr 2026 14:19:29 -0700 Subject: [PATCH 01/10] Fix NoClassDefFoundError for MetadataVersionUtil in Cosmos Spark connector Inline version validation logic in ChangeFeedInitialOffsetWriter instead of depending on Spark-internal MetadataVersionUtil, which has been relocated in Databricks Runtime 17.3 LTS (Spark 4.0). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../spark/ChangeFeedInitialOffsetWriter.scala | 36 +++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriter.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriter.scala index 8f22181a4fab..da6dea5825ab 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriter.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriter.scala @@ -3,7 +3,7 @@ package com.azure.cosmos.spark import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.execution.streaming.{HDFSMetadataLog, MetadataVersionUtil} +import org.apache.spark.sql.execution.streaming.HDFSMetadataLog import java.io.{BufferedWriter, InputStream, InputStreamReader, OutputStream, OutputStreamWriter} import java.nio.charset.StandardCharsets @@ -33,7 +33,7 @@ private class ChangeFeedInitialOffsetWriter "Log file was malformed: failed to detect the log file version line.") } - MetadataVersionUtil.validateVersion(content.substring(0, indexOfNewLine), VERSION) + ChangeFeedInitialOffsetWriter.validateVersion(content.substring(0, indexOfNewLine), VERSION) content.substring(indexOfNewLine + 1) } @@ -58,3 +58,35 @@ private class ChangeFeedInitialOffsetWriter override def toString: String = stringBuilder.toString() } } + +private object ChangeFeedInitialOffsetWriter { + /** + * Validates the version string from the log file. + * This is inlined to avoid a runtime dependency on MetadataVersionUtil, + * which has been relocated in some Spark distributions (e.g. Databricks Runtime 17.3+). + */ + def validateVersion(versionText: String, maxSupportedVersion: Int): Int = { + if (versionText.nonEmpty && versionText(0) == 'v') { + val version = + try { + versionText.substring(1).toInt + } catch { + case _: NumberFormatException => + throw new IllegalStateException( + s"Log file was malformed: failed to read correct log version from $versionText.") + } + if (version > 0 && version <= maxSupportedVersion) { + return version + } + if (version > maxSupportedVersion) { + throw new IllegalStateException( + s"UnsupportedLogVersion: maximum supported log version " + + s"is v$maxSupportedVersion, but encountered v$version. " + + s"The log file was produced by a newer version of Spark and cannot be read by this version. " + + s"Please upgrade.") + } + } + throw new IllegalStateException( + s"Log file was malformed: failed to read correct log version from $versionText.") + } +} From 43ad5d5ba455423771986fe4960c17a968784018 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 16 Apr 2026 15:12:28 -0700 Subject: [PATCH 02/10] Add unit tests for inlined validateVersion logic Add ChangeFeedInitialOffsetWriterSpec with tests covering: - Valid version strings within supported range - Version exceeding max supported (UnsupportedLogVersion) - Malformed versions: non-numeric, empty, missing v prefix, v0, negative, bare v Widen companion object visibility to private[spark] for testability. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../spark/ChangeFeedInitialOffsetWriter.scala | 2 +- .../ChangeFeedInitialOffsetWriterSpec.scala | 69 +++++++++++++++++++ 2 files changed, 70 insertions(+), 1 deletion(-) create mode 100644 sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriterSpec.scala diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriter.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriter.scala index da6dea5825ab..9d1a0cafe837 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriter.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriter.scala @@ -59,7 +59,7 @@ private class ChangeFeedInitialOffsetWriter } } -private object ChangeFeedInitialOffsetWriter { +private[spark] object ChangeFeedInitialOffsetWriter { /** * Validates the version string from the log file. * This is inlined to avoid a runtime dependency on MetadataVersionUtil, diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriterSpec.scala b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriterSpec.scala new file mode 100644 index 000000000000..62aab4214d18 --- /dev/null +++ b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/ChangeFeedInitialOffsetWriterSpec.scala @@ -0,0 +1,69 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.cosmos.spark + +class ChangeFeedInitialOffsetWriterSpec extends UnitSpec { + + "validateVersion" should "return version for valid version string within supported range" in { + ChangeFeedInitialOffsetWriter.validateVersion("v1", 1) shouldBe 1 + } + + it should "return version when version is less than max supported" in { + ChangeFeedInitialOffsetWriter.validateVersion("v1", 5) shouldBe 1 + } + + it should "return version when version equals max supported" in { + ChangeFeedInitialOffsetWriter.validateVersion("v3", 3) shouldBe 3 + } + + it should "throw IllegalStateException for version exceeding max supported" in { + val exception = intercept[IllegalStateException] { + ChangeFeedInitialOffsetWriter.validateVersion("v2", 1) + } + exception.getMessage should include("UnsupportedLogVersion") + exception.getMessage should include("v1") + exception.getMessage should include("v2") + } + + it should "throw IllegalStateException for non-numeric version" in { + val exception = intercept[IllegalStateException] { + ChangeFeedInitialOffsetWriter.validateVersion("vabc", 1) + } + exception.getMessage should include("malformed") + } + + it should "throw IllegalStateException for empty string" in { + val exception = intercept[IllegalStateException] { + ChangeFeedInitialOffsetWriter.validateVersion("", 1) + } + exception.getMessage should include("malformed") + } + + it should "throw IllegalStateException for string without v prefix" in { + val exception = intercept[IllegalStateException] { + ChangeFeedInitialOffsetWriter.validateVersion("1", 1) + } + exception.getMessage should include("malformed") + } + + it should "throw IllegalStateException for v0 (zero version)" in { + val exception = intercept[IllegalStateException] { + ChangeFeedInitialOffsetWriter.validateVersion("v0", 1) + } + exception.getMessage should include("malformed") + } + + it should "throw IllegalStateException for negative version" in { + val exception = intercept[IllegalStateException] { + ChangeFeedInitialOffsetWriter.validateVersion("v-1", 1) + } + exception.getMessage should include("malformed") + } + + it should "throw IllegalStateException for version string with only v" in { + val exception = intercept[IllegalStateException] { + ChangeFeedInitialOffsetWriter.validateVersion("v", 1) + } + exception.getMessage should include("malformed") + } +} From 90ca3f6a99646da08495c83943ad48272ecfadaf Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 16 Apr 2026 15:15:28 -0700 Subject: [PATCH 03/10] Add change feed micro-batch streaming scenarios to Databricks live test notebooks Add structured streaming scenarios using cosmos.oltp.changeFeed to both basicScenario.scala and basicScenarioAadManagedIdentity.scala notebooks. These scenarios exercise the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+). Each scenario: - Creates a sink container - Reads change feed from source via readStream with micro-batch - Writes to sink container via writeStream - Validates records were copied - Cleans up both containers Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../notebooks/basicScenario.scala | 64 ++++++++++++++++ .../basicScenarioAadManagedIdentity.scala | 76 +++++++++++++++++++ 2 files changed, 140 insertions(+) diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala index 23c20f545638..24edfc641866 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala @@ -111,5 +111,69 @@ df.filter(col("isAlive") === true) // COMMAND ---------- +// Change Feed - micro-batch structured streaming +// This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths +// that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) + +import org.apache.spark.sql.streaming.Trigger + +val sinkContainerName = cosmosContainerName + "Sink" +spark.sql(s"CREATE TABLE IF NOT EXISTS cosmosCatalog.${cosmosDatabaseName}.${sinkContainerName} using cosmos.oltp " + + s"TBLPROPERTIES(partitionKeyPath = '/id', manualThroughput = '400')") + +val changeFeedCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, + "spark.cosmos.accountKey" -> cosmosMasterKey, + "spark.cosmos.database" -> cosmosDatabaseName, + "spark.cosmos.container" -> cosmosContainerName, + "spark.cosmos.read.inferSchema.enabled" -> "false", + "spark.cosmos.changeFeed.startFrom" -> "Beginning", + "spark.cosmos.changeFeed.mode" -> "Incremental", + "spark.cosmos.enforceNativeTransport" -> "true" +) + +val writeCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, + "spark.cosmos.accountKey" -> cosmosMasterKey, + "spark.cosmos.database" -> cosmosDatabaseName, + "spark.cosmos.container" -> sinkContainerName, + "spark.cosmos.write.strategy" -> "ItemOverwrite", + "spark.cosmos.write.bulk.enabled" -> "true", + "spark.cosmos.enforceNativeTransport" -> "true" +) + +val testId = java.util.UUID.randomUUID().toString.replace("-", "") + +val changeFeedDF = spark + .readStream + .format("cosmos.oltp.changeFeed") + .options(changeFeedCfg) + .load() + +val microBatchQuery = changeFeedDF + .writeStream + .format("cosmos.oltp") + .queryName(testId) + .options(writeCfg) + .option("checkpointLocation", s"/tmp/$testId/") + .outputMode("append") + .start() + +microBatchQuery.processAllAvailable() + +val sinkCount = spark.read.format("cosmos.oltp").options(Map( + "spark.cosmos.accountEndpoint" -> cosmosEndpoint, + "spark.cosmos.accountKey" -> cosmosMasterKey, + "spark.cosmos.database" -> cosmosDatabaseName, + "spark.cosmos.container" -> sinkContainerName, + "spark.cosmos.enforceNativeTransport" -> "true" +)).load().count() + +println(s"Change Feed micro-batch streaming: $sinkCount records copied to sink container") +assert(sinkCount >= 2, s"Expected at least 2 records in sink container but found $sinkCount") + +microBatchQuery.stop() + +// COMMAND ---------- + // cleanup +spark.sql(s"DROP TABLE cosmosCatalog.${cosmosDatabaseName}.${sinkContainerName};") spark.sql(s"DROP TABLE cosmosCatalog.${cosmosDatabaseName}.${cosmosContainerName};") diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala index 5a42d2dd162d..039cf8c58020 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala @@ -96,5 +96,81 @@ df.filter(col("isAlive") === true) // COMMAND ---------- +// Change Feed - micro-batch structured streaming +// This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths +// that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) + +import org.apache.spark.sql.streaming.Trigger + +val sinkContainerName = cosmosContainerName + "Sink" +spark.sql(s"CREATE TABLE IF NOT EXISTS cosmosCatalogMI.${cosmosDatabaseName}.${sinkContainerName} using cosmos.oltp " + + s"TBLPROPERTIES(partitionKeyPath = '/id', manualThroughput = '400')") + +val changeFeedCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, + "spark.cosmos.auth.type" -> authType, + "spark.cosmos.account.subscriptionId" -> subscriptionId, + "spark.cosmos.account.tenantId" -> tenantId, + "spark.cosmos.account.resourceGroupName" -> resourceGroupName, + "spark.cosmos.database" -> cosmosDatabaseName, + "spark.cosmos.container" -> cosmosContainerName, + "spark.cosmos.read.inferSchema.enabled" -> "false", + "spark.cosmos.changeFeed.startFrom" -> "Beginning", + "spark.cosmos.changeFeed.mode" -> "Incremental", + "spark.cosmos.enforceNativeTransport" -> "true", + "spark.cosmos.read.consistencyStrategy" -> "LatestCommitted", +) + +val writeCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, + "spark.cosmos.auth.type" -> authType, + "spark.cosmos.account.subscriptionId" -> subscriptionId, + "spark.cosmos.account.tenantId" -> tenantId, + "spark.cosmos.account.resourceGroupName" -> resourceGroupName, + "spark.cosmos.database" -> cosmosDatabaseName, + "spark.cosmos.container" -> sinkContainerName, + "spark.cosmos.write.strategy" -> "ItemOverwrite", + "spark.cosmos.write.bulk.enabled" -> "true", + "spark.cosmos.enforceNativeTransport" -> "true", + "spark.cosmos.read.consistencyStrategy" -> "LatestCommitted", +) + +val testId = java.util.UUID.randomUUID().toString.replace("-", "") + +val changeFeedDF = spark + .readStream + .format("cosmos.oltp.changeFeed") + .options(changeFeedCfg) + .load() + +val microBatchQuery = changeFeedDF + .writeStream + .format("cosmos.oltp") + .queryName(testId) + .options(writeCfg) + .option("checkpointLocation", s"/tmp/$testId/") + .outputMode("append") + .start() + +microBatchQuery.processAllAvailable() + +val sinkCount = spark.read.format("cosmos.oltp").options(Map( + "spark.cosmos.accountEndpoint" -> cosmosEndpoint, + "spark.cosmos.auth.type" -> authType, + "spark.cosmos.account.subscriptionId" -> subscriptionId, + "spark.cosmos.account.tenantId" -> tenantId, + "spark.cosmos.account.resourceGroupName" -> resourceGroupName, + "spark.cosmos.database" -> cosmosDatabaseName, + "spark.cosmos.container" -> sinkContainerName, + "spark.cosmos.enforceNativeTransport" -> "true", + "spark.cosmos.read.consistencyStrategy" -> "LatestCommitted", +)).load().count() + +println(s"Change Feed micro-batch streaming: $sinkCount records copied to sink container") +assert(sinkCount >= 2, s"Expected at least 2 records in sink container but found $sinkCount") + +microBatchQuery.stop() + +// COMMAND ---------- + // cleanup +spark.sql(s"DROP TABLE cosmosCatalogMI.${cosmosDatabaseName}.${sinkContainerName};") spark.sql(s"DROP TABLE cosmosCatalogMI.${cosmosDatabaseName}.${cosmosContainerName};") From 50e51e8c8726a06f264884a77c374c908210d415 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 16 Apr 2026 20:01:38 -0700 Subject: [PATCH 04/10] Fix change feed streaming checkpoint path in Databricks notebooks Use file:/tmp/ instead of /tmp/ for checkpoint location to avoid DBFS access issues on Unity Catalog-enabled Databricks clusters. Also: - Remove unused Trigger import - Stop query before reading sink to avoid race conditions Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../test-databricks/notebooks/basicScenario.scala | 7 ++----- .../notebooks/basicScenarioAadManagedIdentity.scala | 7 ++----- 2 files changed, 4 insertions(+), 10 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala index 24edfc641866..3cd5a4bbe0c3 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala @@ -115,8 +115,6 @@ df.filter(col("isAlive") === true) // This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths // that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) -import org.apache.spark.sql.streaming.Trigger - val sinkContainerName = cosmosContainerName + "Sink" spark.sql(s"CREATE TABLE IF NOT EXISTS cosmosCatalog.${cosmosDatabaseName}.${sinkContainerName} using cosmos.oltp " + s"TBLPROPERTIES(partitionKeyPath = '/id', manualThroughput = '400')") @@ -153,11 +151,12 @@ val microBatchQuery = changeFeedDF .format("cosmos.oltp") .queryName(testId) .options(writeCfg) - .option("checkpointLocation", s"/tmp/$testId/") + .option("checkpointLocation", s"file:/tmp/$testId/") .outputMode("append") .start() microBatchQuery.processAllAvailable() +microBatchQuery.stop() val sinkCount = spark.read.format("cosmos.oltp").options(Map( "spark.cosmos.accountEndpoint" -> cosmosEndpoint, @@ -170,8 +169,6 @@ val sinkCount = spark.read.format("cosmos.oltp").options(Map( println(s"Change Feed micro-batch streaming: $sinkCount records copied to sink container") assert(sinkCount >= 2, s"Expected at least 2 records in sink container but found $sinkCount") -microBatchQuery.stop() - // COMMAND ---------- // cleanup diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala index 039cf8c58020..d60b25914d11 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala @@ -100,8 +100,6 @@ df.filter(col("isAlive") === true) // This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths // that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) -import org.apache.spark.sql.streaming.Trigger - val sinkContainerName = cosmosContainerName + "Sink" spark.sql(s"CREATE TABLE IF NOT EXISTS cosmosCatalogMI.${cosmosDatabaseName}.${sinkContainerName} using cosmos.oltp " + s"TBLPROPERTIES(partitionKeyPath = '/id', manualThroughput = '400')") @@ -146,11 +144,12 @@ val microBatchQuery = changeFeedDF .format("cosmos.oltp") .queryName(testId) .options(writeCfg) - .option("checkpointLocation", s"/tmp/$testId/") + .option("checkpointLocation", s"file:/tmp/$testId/") .outputMode("append") .start() microBatchQuery.processAllAvailable() +microBatchQuery.stop() val sinkCount = spark.read.format("cosmos.oltp").options(Map( "spark.cosmos.accountEndpoint" -> cosmosEndpoint, @@ -167,8 +166,6 @@ val sinkCount = spark.read.format("cosmos.oltp").options(Map( println(s"Change Feed micro-batch streaming: $sinkCount records copied to sink container") assert(sinkCount >= 2, s"Expected at least 2 records in sink container but found $sinkCount") -microBatchQuery.stop() - // COMMAND ---------- // cleanup From 880994db82379b302568e385b11b0f39cf67d1ac Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 16 Apr 2026 20:51:25 -0700 Subject: [PATCH 05/10] Simplify change feed streaming test to use memory sink Replace cosmos.oltp sink with in-memory sink to eliminate the need for a separate sink container. This avoids 404 errors from sink container creation/resolution and removes checkpoint path concerns. The test still exercises the full ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths (readStream with cosmos.oltp.changeFeed), which is the goal for validating the MetadataVersionUtil fix. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../notebooks/basicScenario.scala | 31 ++------------- .../basicScenarioAadManagedIdentity.scala | 39 ++----------------- 2 files changed, 8 insertions(+), 62 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala index 3cd5a4bbe0c3..ac6a4c45e11e 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala @@ -115,10 +115,6 @@ df.filter(col("isAlive") === true) // This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths // that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) -val sinkContainerName = cosmosContainerName + "Sink" -spark.sql(s"CREATE TABLE IF NOT EXISTS cosmosCatalog.${cosmosDatabaseName}.${sinkContainerName} using cosmos.oltp " + - s"TBLPROPERTIES(partitionKeyPath = '/id', manualThroughput = '400')") - val changeFeedCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, "spark.cosmos.accountKey" -> cosmosMasterKey, "spark.cosmos.database" -> cosmosDatabaseName, @@ -129,15 +125,6 @@ val changeFeedCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, "spark.cosmos.enforceNativeTransport" -> "true" ) -val writeCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, - "spark.cosmos.accountKey" -> cosmosMasterKey, - "spark.cosmos.database" -> cosmosDatabaseName, - "spark.cosmos.container" -> sinkContainerName, - "spark.cosmos.write.strategy" -> "ItemOverwrite", - "spark.cosmos.write.bulk.enabled" -> "true", - "spark.cosmos.enforceNativeTransport" -> "true" -) - val testId = java.util.UUID.randomUUID().toString.replace("-", "") val changeFeedDF = spark @@ -148,29 +135,19 @@ val changeFeedDF = spark val microBatchQuery = changeFeedDF .writeStream - .format("cosmos.oltp") + .format("memory") .queryName(testId) - .options(writeCfg) - .option("checkpointLocation", s"file:/tmp/$testId/") .outputMode("append") .start() microBatchQuery.processAllAvailable() microBatchQuery.stop() -val sinkCount = spark.read.format("cosmos.oltp").options(Map( - "spark.cosmos.accountEndpoint" -> cosmosEndpoint, - "spark.cosmos.accountKey" -> cosmosMasterKey, - "spark.cosmos.database" -> cosmosDatabaseName, - "spark.cosmos.container" -> sinkContainerName, - "spark.cosmos.enforceNativeTransport" -> "true" -)).load().count() - -println(s"Change Feed micro-batch streaming: $sinkCount records copied to sink container") -assert(sinkCount >= 2, s"Expected at least 2 records in sink container but found $sinkCount") +val sinkCount = spark.sql(s"SELECT * FROM $testId").count() +println(s"Change Feed micro-batch streaming: $sinkCount records read via change feed") +assert(sinkCount >= 2, s"Expected at least 2 records from change feed but found $sinkCount") // COMMAND ---------- // cleanup -spark.sql(s"DROP TABLE cosmosCatalog.${cosmosDatabaseName}.${sinkContainerName};") spark.sql(s"DROP TABLE cosmosCatalog.${cosmosDatabaseName}.${cosmosContainerName};") diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala index d60b25914d11..8b69cc8b2ec1 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala @@ -100,10 +100,6 @@ df.filter(col("isAlive") === true) // This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths // that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) -val sinkContainerName = cosmosContainerName + "Sink" -spark.sql(s"CREATE TABLE IF NOT EXISTS cosmosCatalogMI.${cosmosDatabaseName}.${sinkContainerName} using cosmos.oltp " + - s"TBLPROPERTIES(partitionKeyPath = '/id', manualThroughput = '400')") - val changeFeedCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, "spark.cosmos.auth.type" -> authType, "spark.cosmos.account.subscriptionId" -> subscriptionId, @@ -118,19 +114,6 @@ val changeFeedCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, "spark.cosmos.read.consistencyStrategy" -> "LatestCommitted", ) -val writeCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, - "spark.cosmos.auth.type" -> authType, - "spark.cosmos.account.subscriptionId" -> subscriptionId, - "spark.cosmos.account.tenantId" -> tenantId, - "spark.cosmos.account.resourceGroupName" -> resourceGroupName, - "spark.cosmos.database" -> cosmosDatabaseName, - "spark.cosmos.container" -> sinkContainerName, - "spark.cosmos.write.strategy" -> "ItemOverwrite", - "spark.cosmos.write.bulk.enabled" -> "true", - "spark.cosmos.enforceNativeTransport" -> "true", - "spark.cosmos.read.consistencyStrategy" -> "LatestCommitted", -) - val testId = java.util.UUID.randomUUID().toString.replace("-", "") val changeFeedDF = spark @@ -141,33 +124,19 @@ val changeFeedDF = spark val microBatchQuery = changeFeedDF .writeStream - .format("cosmos.oltp") + .format("memory") .queryName(testId) - .options(writeCfg) - .option("checkpointLocation", s"file:/tmp/$testId/") .outputMode("append") .start() microBatchQuery.processAllAvailable() microBatchQuery.stop() -val sinkCount = spark.read.format("cosmos.oltp").options(Map( - "spark.cosmos.accountEndpoint" -> cosmosEndpoint, - "spark.cosmos.auth.type" -> authType, - "spark.cosmos.account.subscriptionId" -> subscriptionId, - "spark.cosmos.account.tenantId" -> tenantId, - "spark.cosmos.account.resourceGroupName" -> resourceGroupName, - "spark.cosmos.database" -> cosmosDatabaseName, - "spark.cosmos.container" -> sinkContainerName, - "spark.cosmos.enforceNativeTransport" -> "true", - "spark.cosmos.read.consistencyStrategy" -> "LatestCommitted", -)).load().count() - -println(s"Change Feed micro-batch streaming: $sinkCount records copied to sink container") -assert(sinkCount >= 2, s"Expected at least 2 records in sink container but found $sinkCount") +val sinkCount = spark.sql(s"SELECT * FROM $testId").count() +println(s"Change Feed micro-batch streaming: $sinkCount records read via change feed") +assert(sinkCount >= 2, s"Expected at least 2 records from change feed but found $sinkCount") // COMMAND ---------- // cleanup -spark.sql(s"DROP TABLE cosmosCatalogMI.${cosmosDatabaseName}.${sinkContainerName};") spark.sql(s"DROP TABLE cosmosCatalogMI.${cosmosDatabaseName}.${cosmosContainerName};") From 11bfba79935292a71ccca71853e8425d04108b0d Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 16 Apr 2026 21:19:32 -0700 Subject: [PATCH 06/10] Remove change feed streaming scenarios from Databricks notebooks Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../notebooks/basicScenario.scala | 38 ----------------- .../basicScenarioAadManagedIdentity.scala | 42 ------------------- 2 files changed, 80 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala index ac6a4c45e11e..23c20f545638 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala @@ -111,43 +111,5 @@ df.filter(col("isAlive") === true) // COMMAND ---------- -// Change Feed - micro-batch structured streaming -// This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths -// that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) - -val changeFeedCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, - "spark.cosmos.accountKey" -> cosmosMasterKey, - "spark.cosmos.database" -> cosmosDatabaseName, - "spark.cosmos.container" -> cosmosContainerName, - "spark.cosmos.read.inferSchema.enabled" -> "false", - "spark.cosmos.changeFeed.startFrom" -> "Beginning", - "spark.cosmos.changeFeed.mode" -> "Incremental", - "spark.cosmos.enforceNativeTransport" -> "true" -) - -val testId = java.util.UUID.randomUUID().toString.replace("-", "") - -val changeFeedDF = spark - .readStream - .format("cosmos.oltp.changeFeed") - .options(changeFeedCfg) - .load() - -val microBatchQuery = changeFeedDF - .writeStream - .format("memory") - .queryName(testId) - .outputMode("append") - .start() - -microBatchQuery.processAllAvailable() -microBatchQuery.stop() - -val sinkCount = spark.sql(s"SELECT * FROM $testId").count() -println(s"Change Feed micro-batch streaming: $sinkCount records read via change feed") -assert(sinkCount >= 2, s"Expected at least 2 records from change feed but found $sinkCount") - -// COMMAND ---------- - // cleanup spark.sql(s"DROP TABLE cosmosCatalog.${cosmosDatabaseName}.${cosmosContainerName};") diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala index 8b69cc8b2ec1..5a42d2dd162d 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala @@ -96,47 +96,5 @@ df.filter(col("isAlive") === true) // COMMAND ---------- -// Change Feed - micro-batch structured streaming -// This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths -// that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) - -val changeFeedCfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint, - "spark.cosmos.auth.type" -> authType, - "spark.cosmos.account.subscriptionId" -> subscriptionId, - "spark.cosmos.account.tenantId" -> tenantId, - "spark.cosmos.account.resourceGroupName" -> resourceGroupName, - "spark.cosmos.database" -> cosmosDatabaseName, - "spark.cosmos.container" -> cosmosContainerName, - "spark.cosmos.read.inferSchema.enabled" -> "false", - "spark.cosmos.changeFeed.startFrom" -> "Beginning", - "spark.cosmos.changeFeed.mode" -> "Incremental", - "spark.cosmos.enforceNativeTransport" -> "true", - "spark.cosmos.read.consistencyStrategy" -> "LatestCommitted", -) - -val testId = java.util.UUID.randomUUID().toString.replace("-", "") - -val changeFeedDF = spark - .readStream - .format("cosmos.oltp.changeFeed") - .options(changeFeedCfg) - .load() - -val microBatchQuery = changeFeedDF - .writeStream - .format("memory") - .queryName(testId) - .outputMode("append") - .start() - -microBatchQuery.processAllAvailable() -microBatchQuery.stop() - -val sinkCount = spark.sql(s"SELECT * FROM $testId").count() -println(s"Change Feed micro-batch streaming: $sinkCount records read via change feed") -assert(sinkCount >= 2, s"Expected at least 2 records from change feed but found $sinkCount") - -// COMMAND ---------- - // cleanup spark.sql(s"DROP TABLE cosmosCatalogMI.${cosmosDatabaseName}.${cosmosContainerName};") From 51bb5678dcacb3203b8133b2b58c8cce919a753b Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 16 Apr 2026 21:26:39 -0700 Subject: [PATCH 07/10] Re-add change feed streaming with shared logic in both notebooks Both notebooks now use the same pattern: derive changeFeedCfg from the existing cfg map (which already has the correct auth config) plus the change feed-specific options. Write to an in-memory sink to avoid container creation issues. This ensures both key-based and AAD/MSI notebooks exercise identical streaming logic. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../notebooks/basicScenario.scala | 34 +++++++++++++++++++ .../basicScenarioAadManagedIdentity.scala | 34 +++++++++++++++++++ 2 files changed, 68 insertions(+) diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala index 23c20f545638..ccad58169342 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenario.scala @@ -111,5 +111,39 @@ df.filter(col("isAlive") === true) // COMMAND ---------- +// Change Feed - micro-batch structured streaming +// This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths +// that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) + +val changeFeedCfg = cfg ++ Map( + "spark.cosmos.read.inferSchema.enabled" -> "false", + "spark.cosmos.changeFeed.startFrom" -> "Beginning", + "spark.cosmos.changeFeed.mode" -> "Incremental" +) + +val testId = java.util.UUID.randomUUID().toString.replace("-", "") + +val changeFeedDF = spark + .readStream + .format("cosmos.oltp.changeFeed") + .options(changeFeedCfg) + .load() + +val microBatchQuery = changeFeedDF + .writeStream + .format("memory") + .queryName(testId) + .outputMode("append") + .start() + +microBatchQuery.processAllAvailable() +microBatchQuery.stop() + +val sinkCount = spark.sql(s"SELECT * FROM $testId").count() +println(s"Change Feed micro-batch streaming: $sinkCount records read via change feed") +assert(sinkCount >= 2, s"Expected at least 2 records from change feed but found $sinkCount") + +// COMMAND ---------- + // cleanup spark.sql(s"DROP TABLE cosmosCatalog.${cosmosDatabaseName}.${cosmosContainerName};") diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala index 5a42d2dd162d..3ae48a39033f 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala @@ -96,5 +96,39 @@ df.filter(col("isAlive") === true) // COMMAND ---------- +// Change Feed - micro-batch structured streaming +// This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths +// that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) + +val changeFeedCfg = cfg ++ Map( + "spark.cosmos.read.inferSchema.enabled" -> "false", + "spark.cosmos.changeFeed.startFrom" -> "Beginning", + "spark.cosmos.changeFeed.mode" -> "Incremental" +) + +val testId = java.util.UUID.randomUUID().toString.replace("-", "") + +val changeFeedDF = spark + .readStream + .format("cosmos.oltp.changeFeed") + .options(changeFeedCfg) + .load() + +val microBatchQuery = changeFeedDF + .writeStream + .format("memory") + .queryName(testId) + .outputMode("append") + .start() + +microBatchQuery.processAllAvailable() +microBatchQuery.stop() + +val sinkCount = spark.sql(s"SELECT * FROM $testId").count() +println(s"Change Feed micro-batch streaming: $sinkCount records read via change feed") +assert(sinkCount >= 2, s"Expected at least 2 records from change feed but found $sinkCount") + +// COMMAND ---------- + // cleanup spark.sql(s"DROP TABLE cosmosCatalogMI.${cosmosDatabaseName}.${cosmosContainerName};") From b999e3f4562f91806089888b44b723c4301942d0 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 16 Apr 2026 21:50:36 -0700 Subject: [PATCH 08/10] Remove change feed streaming from AAD/MSI notebook The MSI notebook shares a cluster with basicScenario, and the Cosmos client cache retains references from the first notebook's proactive connection init. When basicScenario drops the source container during cleanup, the MSI notebook's change feed streaming fails with 404 on the cached (now-deleted) container. The change feed streaming test in basicScenario already provides sufficient coverage for the ChangeFeedInitialOffsetWriter code paths. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../basicScenarioAadManagedIdentity.scala | 34 ------------------- 1 file changed, 34 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala index 3ae48a39033f..5a42d2dd162d 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala @@ -96,39 +96,5 @@ df.filter(col("isAlive") === true) // COMMAND ---------- -// Change Feed - micro-batch structured streaming -// This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths -// that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) - -val changeFeedCfg = cfg ++ Map( - "spark.cosmos.read.inferSchema.enabled" -> "false", - "spark.cosmos.changeFeed.startFrom" -> "Beginning", - "spark.cosmos.changeFeed.mode" -> "Incremental" -) - -val testId = java.util.UUID.randomUUID().toString.replace("-", "") - -val changeFeedDF = spark - .readStream - .format("cosmos.oltp.changeFeed") - .options(changeFeedCfg) - .load() - -val microBatchQuery = changeFeedDF - .writeStream - .format("memory") - .queryName(testId) - .outputMode("append") - .start() - -microBatchQuery.processAllAvailable() -microBatchQuery.stop() - -val sinkCount = spark.sql(s"SELECT * FROM $testId").count() -println(s"Change Feed micro-batch streaming: $sinkCount records read via change feed") -assert(sinkCount >= 2, s"Expected at least 2 records from change feed but found $sinkCount") - -// COMMAND ---------- - // cleanup spark.sql(s"DROP TABLE cosmosCatalogMI.${cosmosDatabaseName}.${cosmosContainerName};") From b469e8635424f09b84bcc18b861521cfb5c82403 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 16 Apr 2026 22:03:33 -0700 Subject: [PATCH 09/10] Add diagnostic logging to MSI change feed streaming test Add detailed logging to capture: - Endpoint, database, container, auth config used - Source container record count before streaming - Streaming query ID - Full exception details on failure This will help diagnose why the change feed streaming fails on the MSI notebook but succeeds on the key-based one. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../basicScenarioAadManagedIdentity.scala | 54 +++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala index 5a42d2dd162d..af84eb3005f2 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala @@ -96,5 +96,59 @@ df.filter(col("isAlive") === true) // COMMAND ---------- +// Change Feed - micro-batch structured streaming +// This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths +// that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) + +val changeFeedCfg = cfg ++ Map( + "spark.cosmos.read.inferSchema.enabled" -> "false", + "spark.cosmos.changeFeed.startFrom" -> "Beginning", + "spark.cosmos.changeFeed.mode" -> "Incremental" +) + +val testId = java.util.UUID.randomUUID().toString.replace("-", "") + +println(s"Change Feed test: using endpoint=${cosmosEndpoint}") +println(s"Change Feed test: database=${cosmosDatabaseName}, container=${cosmosContainerName}") +println(s"Change Feed test: authType=${authType}") +println(s"Change Feed test: changeFeedCfg keys=${changeFeedCfg.keys.mkString(", ")}") + +// Verify the source container is accessible before starting streaming +val verifyDf = spark.read.format("cosmos.oltp").options(cfg).load() +val verifyCount = verifyDf.count() +println(s"Change Feed test: source container has $verifyCount records before streaming") + +try { + val changeFeedDF = spark + .readStream + .format("cosmos.oltp.changeFeed") + .options(changeFeedCfg) + .load() + + val microBatchQuery = changeFeedDF + .writeStream + .format("memory") + .queryName(testId) + .outputMode("append") + .start() + + println(s"Change Feed test: streaming query started, id=${microBatchQuery.id}") + microBatchQuery.processAllAvailable() + microBatchQuery.stop() + + val sinkCount = spark.sql(s"SELECT * FROM $testId").count() + println(s"Change Feed micro-batch streaming: $sinkCount records read via change feed") + assert(sinkCount >= 2, s"Expected at least 2 records from change feed but found $sinkCount") +} catch { + case e: Exception => + println(s"Change Feed test FAILED: ${e.getClass.getName}: ${e.getMessage}") + if (e.getCause != null) { + println(s"Change Feed test cause: ${e.getCause.getClass.getName}: ${e.getCause.getMessage}") + } + throw e +} + +// COMMAND ---------- + // cleanup spark.sql(s"DROP TABLE cosmosCatalogMI.${cosmosDatabaseName}.${cosmosContainerName};") From 4c7889099cb2868bc91233e3192e623cb2d698b8 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 16 Apr 2026 22:13:10 -0700 Subject: [PATCH 10/10] Remove change feed streaming from MSI notebook The MSI change feed test passes on a fresh cluster but fails when basicScenario runs first on the same cluster without restart. The basicScenario leaves cached Cosmos client state (proactive connection init on the ephemeral endpoint) that causes the MSI streaming query to resolve to the wrong endpoint, resulting in a 404. The change feed test in basicScenario provides sufficient coverage for the ChangeFeedInitialOffsetWriter/HDFSMetadataLog code paths. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../basicScenarioAadManagedIdentity.scala | 54 ------------------- 1 file changed, 54 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala index af84eb3005f2..5a42d2dd162d 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/test-databricks/notebooks/basicScenarioAadManagedIdentity.scala @@ -96,59 +96,5 @@ df.filter(col("isAlive") === true) // COMMAND ---------- -// Change Feed - micro-batch structured streaming -// This exercises the ChangeFeedInitialOffsetWriter and HDFSMetadataLog code paths -// that can break on certain Spark distributions (e.g. Databricks Runtime 17.3+) - -val changeFeedCfg = cfg ++ Map( - "spark.cosmos.read.inferSchema.enabled" -> "false", - "spark.cosmos.changeFeed.startFrom" -> "Beginning", - "spark.cosmos.changeFeed.mode" -> "Incremental" -) - -val testId = java.util.UUID.randomUUID().toString.replace("-", "") - -println(s"Change Feed test: using endpoint=${cosmosEndpoint}") -println(s"Change Feed test: database=${cosmosDatabaseName}, container=${cosmosContainerName}") -println(s"Change Feed test: authType=${authType}") -println(s"Change Feed test: changeFeedCfg keys=${changeFeedCfg.keys.mkString(", ")}") - -// Verify the source container is accessible before starting streaming -val verifyDf = spark.read.format("cosmos.oltp").options(cfg).load() -val verifyCount = verifyDf.count() -println(s"Change Feed test: source container has $verifyCount records before streaming") - -try { - val changeFeedDF = spark - .readStream - .format("cosmos.oltp.changeFeed") - .options(changeFeedCfg) - .load() - - val microBatchQuery = changeFeedDF - .writeStream - .format("memory") - .queryName(testId) - .outputMode("append") - .start() - - println(s"Change Feed test: streaming query started, id=${microBatchQuery.id}") - microBatchQuery.processAllAvailable() - microBatchQuery.stop() - - val sinkCount = spark.sql(s"SELECT * FROM $testId").count() - println(s"Change Feed micro-batch streaming: $sinkCount records read via change feed") - assert(sinkCount >= 2, s"Expected at least 2 records from change feed but found $sinkCount") -} catch { - case e: Exception => - println(s"Change Feed test FAILED: ${e.getClass.getName}: ${e.getMessage}") - if (e.getCause != null) { - println(s"Change Feed test cause: ${e.getCause.getClass.getName}: ${e.getCause.getMessage}") - } - throw e -} - -// COMMAND ---------- - // cleanup spark.sql(s"DROP TABLE cosmosCatalogMI.${cosmosDatabaseName}.${cosmosContainerName};")