Add ReadConsistencyStrategy API (Azure#45161)

* Add ReadConsistencyStrategy API

* Adding test coverage

* Update ClientConfigDiagnosticsTest.java

* Update ConsistencyReaderTest.java

* Update SessionTest.java

* Update OperationPoliciesTest.java

* Update OperationPoliciesTest.java

* Update ReadConsistencyStrategy.java

* Reacted to code review feedback

* Update sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/models/CosmosChangeFeedRequestOptions.java

Co-authored-by: Abhijeet Mohanty <[email protected]>

* Update ReadConsistencyStrategy.java

---------

Co-authored-by: Abhijeet Mohanty <[email protected]>

Author: FabianMeiswinkel

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

@@ -37,7 +37,7 @@ private class ChangeFeedMicroBatchStream
   private val readConfig = CosmosReadConfig.parseCosmosReadConfig(config)
   private val sparkEnvironmentInfo = CosmosClientConfiguration.getSparkEnvironmentInfo(Some(session))
   private val clientConfiguration = CosmosClientConfiguration.apply(
-    config, readConfig.forceEventualConsistency, sparkEnvironmentInfo)
+    config, readConfig.readConsistencyStrategy, sparkEnvironmentInfo)
   private val containerConfig = CosmosContainerConfig.parseCosmosContainerConfig(config)
   private val partitioningConfig = CosmosPartitioningConfig.parseCosmosPartitioningConfig(config)
   private val changeFeedConfig = CosmosChangeFeedConfig.parseCosmosChangeFeedConfig(config)

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

@@ -35,7 +35,7 @@ private case class ItemsScanBuilder(session: SparkSession,
   private var processedPredicates : Option[AnalyzedAggregatedFilters] = Option.empty
   private val clientConfiguration = CosmosClientConfiguration.apply(
     configMap,
-    readConfig.forceEventualConsistency,
+    readConfig.readConsistencyStrategy,
     CosmosClientConfiguration.getSparkEnvironmentInfo(Some(session))
   )
   private val containerConfig = CosmosContainerConfig.parseCosmosContainerConfig(configMap)

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

@@ -38,7 +38,7 @@ private class ChangeFeedMicroBatchStream
   private val sparkEnvironmentInfo = CosmosClientConfiguration.getSparkEnvironmentInfo(Some(session))
   private val clientConfiguration = CosmosClientConfiguration.apply(
     config,
-    readConfig.forceEventualConsistency,
+    readConfig.readConsistencyStrategy,
     sparkEnvironmentInfo)
   private val containerConfig = CosmosContainerConfig.parseCosmosContainerConfig(config)
   private val partitioningConfig = CosmosPartitioningConfig.parseCosmosPartitioningConfig(config)

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

@@ -36,7 +36,7 @@ private case class ItemsScanBuilder(session: SparkSession,
 
   private val clientConfiguration = CosmosClientConfiguration.apply(
     configMap,
-    readConfig.forceEventualConsistency,
+    readConfig.readConsistencyStrategy,
     CosmosClientConfiguration.getSparkEnvironmentInfo(Some(session))
   )
   private val containerConfig = CosmosContainerConfig.parseCosmosContainerConfig(configMap)

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

@@ -38,7 +38,7 @@ private class ChangeFeedMicroBatchStream
   private val sparkEnvironmentInfo = CosmosClientConfiguration.getSparkEnvironmentInfo(Some(session))
   private val clientConfiguration = CosmosClientConfiguration.apply(
     config,
-    readConfig.forceEventualConsistency,
+    readConfig.readConsistencyStrategy,
     sparkEnvironmentInfo)
   private val containerConfig = CosmosContainerConfig.parseCosmosContainerConfig(config)
   private val partitioningConfig = CosmosPartitioningConfig.parseCosmosPartitioningConfig(config)

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

@@ -36,7 +36,7 @@ private case class ItemsScanBuilder(session: SparkSession,
 
   private val clientConfiguration = CosmosClientConfiguration.apply(
     configMap,
-    readConfig.forceEventualConsistency,
+    readConfig.readConsistencyStrategy,
     CosmosClientConfiguration.getSparkEnvironmentInfo(Some(session))
   )
   private val containerConfig = CosmosContainerConfig.parseCosmosContainerConfig(configMap)

sdk/cosmos/azure-cosmos-spark_3_2-12/docs/configuration-reference.md

@@ -28,6 +28,7 @@
 | `spark.cosmos.enforceNativeTransport`                             | `false` | Indicates whether native netty transport should be enforced. If true, on client creation an exception will be thrown if netty is not able to use native transport. The native transport is more efficient especially when there is a high number of connections.                                                                                                                                                                                                                                                                                                                                              |
 | `spark.cosmos.http.connectionPoolSize`                            | `1000`  | Gateway mode connection pool size                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
 | `spark.cosmos.read.forceEventualConsistency`                      | `true`  | Makes the client use Eventual consistency for read operations instead of using the default account level consistency                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
+| `spark.cosmos.read.consistencyStrategy`                           | None    | Overrides the read consistency strategy - will override `spark.cosmos.read.forceEventualConsistency` if both are specified. `EVENTUAL` means the same as setting `spark.cosmos.read.forceEventualConsistency` to `true`. `LOCAL_COMMITTED` will ensure quorum reads, `SESSION` will apply session consistency and `DEFAULT` applies the default account-level consistency. (Same as `spark.cosmos.read.forceEventualConsistency` beign `false`). This API is still in preview - currently only `DEFAULT` and `EVENTUAL` can be used in Gateway mode.                                                          |
 | `spark.cosmos.applicationName`                                    | None    | Application name                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              |
 | `spark.cosmos.preferredRegionsList`                               | None    | Preferred regions list to be used for a multi region Cosmos DB account. This is a comma separated value (e.g., `[East US, West US]` or `East US, West US`) provided preferred regions will be used as hint. You should use a collocated spark cluster with your Cosmos DB account and pass the spark cluster region as preferred region. See list of azure regions [here](https://docs.microsoft.com/dotnet/api/microsoft.azure.documents.locationnames?view=azure-dotnet&preserve-view=true). Please note that you can also use `spark.cosmos.preferredRegions` as alias                                     |
 | `spark.cosmos.diagnostics`                                        | None    | Can be used to enable more verbose diagnostics. Currently the only supported options are `simple`, `feed` and `feed_details`. The mode `simple` will result in additional logs being emitted as `INFO` logs in the Driver and Executor logs. The mode `feed` will provide per `FeedResponse` information and `feed_details` will also add pk and id values of all items in a `FeedResponse` to the logs. Please note that any of the diagnostic modes causes the logs being emitted to increase significantly - so, there will be perf impact when enabling it.                                               |