{"id":7466,"date":"2024-01-18T07:00:01","date_gmt":"2024-01-18T15:00:01","guid":{"rendered":"https:\/\/devblogs.microsoft.com\/cosmosdb\/?p=7466"},"modified":"2024-08-08T07:34:40","modified_gmt":"2024-08-08T14:34:40","slug":"latest-nosql-java-ecosystem-updates-2023-q3-q4","status":"publish","type":"post","link":"https:\/\/devblogs.microsoft.com\/cosmosdb\/latest-nosql-java-ecosystem-updates-2023-q3-q4\/","title":{"rendered":"Latest NoSQL Java Ecosystem Updates 2023 Q3 & Q4"},"content":{"rendered":"

We’re always busy adding new features, fixes, patches, and improvements to our Java-based client libraries for Azure Cosmos DB for NoSQL<\/a>. In this regular blog series, we share highlights of recent updates in the last period.<\/p>\n

 <\/p>\n

July – December 2023 updates<\/h2>\n

 <\/p>\n

    \n
  1. Spark 3.4 Support<\/a><\/li>\n
  2. Throughput Control – gateway support in Spark Connector<\/a><\/li>\n
  3. Aggressive Connection Warmup Improvements in Java SDK<\/a><\/li>\n
  4. Query Pagination Improvements in Java SDK<\/a><\/li>\n
  5. Patch Operation on more than 10 fields in Spark Connector<\/a><\/li>\n
  6. Bypass Integrated Cache in Java SDK<\/a><\/li>\n
  7. Diagnostics Thresholds support for Java SDK and Spring Data<\/a><\/li>\n
  8. Integration of Throughput Control with Change Feed Processor – Java SDK<\/a><\/li>\n
  9. Session token mismatch\u00a0 – further optimization<\/a><\/li>\n
  10. Change Feed Processor Context for All Versions and Deletes mode in Java SDK<\/a><\/li>\n
  11. Hierarchical Partition Key Support in Spark Connector<\/a><\/li>\n<\/ol>\n

     <\/p>\n

    Spark 3.4 Support<\/h5>\n

    Cloud-native hybrid transactional and analytical processing (HTAP)<\/a>\u00a0is supported in Azure Cosmos DB through\u00a0Synapse link<\/a>, using OLTP and OLAP Spark connectors, which now support Spark 3.4 as of July 2023. This includes support for TimestampNTZType<\/a> introduced in Spark 3.4.<\/p>\n

     <\/p>\n

    Throughput Control \u2013 Gateway Mode Support in Spark Connector<\/h5>\n

    In the Cosmos DB Spark Connector<\/a>, throughput control is a feature which helps to isolate the performance needs of applications running against a container, by limiting the amount of request units<\/a> that can be consumed by a given Spark client. In September 2023 we added support for gateway mode<\/a>, users can now enable this by setting spark.cosmos.useGatewayMode<\/code> to true <\/strong>in Spark config. Find out more about throughput control here<\/a>.<\/p>\n

     <\/p>\n

    Aggressive Connection Warmup Improvements in Java SDK<\/h5>\n

    Earlier in February 2023 we introduced Proactive Connection Management<\/a>, a feature which allows developers to warm up connections and caches for containers in both the current read region and a pre-defined number of preferred remote regions. This feature can improve tail latency in cross-region failover scenarios. In September we have made enhancements and improved the efficiency<\/a> of the way connections are opened during the warm up phase.<\/p>\n

     <\/p>\n

    Query Pagination Improvements in Java SDK<\/h5>\n

    We’ve made enhancements\u00a0to pagination<\/a> in September 2023.<\/p>\n

     <\/p>\n

    Patch Operation on more than 10 fields in Spark Connector<\/h5>\n

    For Patch API in Cosmos DB, there is a general limitation of only patching documents up to 10 fields at a time (see Partial document update – Azure Cosmos DB for NoSQL<\/a> for more information). In the Spark OLTP connector we have developed a new feature to allow customers to patch their documents having more than 10 columns. This resolves a big restriction on most of the customers who use spark for bulk ingestion and updating their documents. By setting spark.cosmos.write.strategy<\/code> to\u00a0 ItemBulkUpdate<\/code>, users can now patch more than 10 items in a single operation. To find samples for doing patch using Cosmos DB Spark Connector, see here<\/a>.<\/p>\n

     <\/p>\n

    Bypass Integrated Cache in Java SDK<\/h5>\n

    The Azure Cosmos DB integrated cache<\/a> is an in-memory cache that helps ensure manageable costs and low latency as request volume grows. The integrated cache is a read-through, write-through cache with a Least Recently Used (LRU) eviction policy. However, there are some scenarios where it is preferable to avoid using the cache on a per request basis:<\/p>\n

    DedicatedGatewayRequestOptions dedicatedOptions = new DedicatedGatewayRequestOptions(); \r\ndedicatedOptions.setMaxIntegratedCacheStaleness(Duration.ofMinutes(2)); \r\ndedicatedOptions.setIntegratedCacheBypassed(true); \r\nCosmosQueryRequestOptions queryOptions = new CosmosQueryRequestOptions(); \r\nqueryOptions.setDedicatedGatewayRequestOptions(dedicatedOptions);<\/code><\/pre>\n
    <\/h5>\n
    Diagnostics Thresholds support for Java SDK and Spring Data<\/h5>\n
    Back in September 2022 we introduced the option to emit client metrics<\/a> from the Azure Cosmos DB Java SDK via micrometer MeterRegistry<\/a> as well as doing so from the Spark connector via configuration. We have since added the ability to define thresholds<\/strong>, which for very noisy applications will help to limit metrics consumed to the ones you are most interested in. Check out documentation on instrumenting client metrics with Micrometer using Prometheus<\/a>, which includes examples on how to define thresholds. We also recently enabled threshold support in the Cosmos Spring Data Client Library<\/a>, just configure your CosmosClientBuilder<\/code> bean as below:<\/p>\n
    @Bean\r\npublic CosmosClientBuilder cosmosClientBuilder() {\r\n    DirectConnectionConfig directConnectionConfig = DirectConnectionConfig.getDefaultConfig();\r\n    return new CosmosClientBuilder()\r\n            .endpoint(properties.getUri())\r\n            .key(properties.getKey())\r\n            .directMode(directConnectionConfig)\r\n            .clientTelemetryConfig(\r\n                    new CosmosClientTelemetryConfig()\r\n                            .diagnosticsThresholds(\r\n                                    new CosmosDiagnosticsThresholds()\r\n                                                .setNonPointOperationLatencyThreshold(Duration.ofMillis(nonPointOperationLatencyThresholdInMS))\r\n                                                .setPointOperationLatencyThreshold(Duration.ofMillis(pointOperationLatencyThresholdInMS))\r\n                                                .setPayloadSizeThreshold(payloadSizeThresholdInBytes)\r\n                                                .setRequestChargeThreshold(requestChargeThresholdInRU)\r\n                                )\r\n                                .diagnosticsHandler(CosmosDiagnosticsHandler.DEFAULT_LOGGING_HANDLER));\r\n    }<\/code><\/pre>\n
    <\/h5>\n
    <\/h5>\n
    Integration of Throughput Control with Change Feed Processor – Java SDK<\/h5>\n
    During a heavy backlog of changes in the monitoring container, the Change Feed Processor<\/a> (CFP) will keep polling documents in order to catch up. This can cause an increase in request unit<\/a> usage, and in turn cause heavy throttling. To avoid this, we have integrated Throughput Control with CFP. In Java SDK, Priority Based Execution<\/a> is already integrated into Throughput Control, making it easier for customers to set low priority for CFP based workload to avoid the impact of throttling on other workloads running in parallel. Here’s a sample of how to define both throughput control and a priority level when processing change feed:<\/p>\n
    ThroughputControlGroupConfig throughputControlGroupConfig =\r\n        new ThroughputControlGroupConfigBuilder()\r\n                .groupName(\"changeFeedProcessor\")\r\n                .targetThroughput(1000)\r\n                .priorityLevel(PriorityLevel.LOW)\r\n                .build();\r\noptions.setFeedPollThroughputControlConfig(throughputControlGroupConfig);\r\n\r\nChangeFeedProcessor changeFeedProcessorInstance = new ChangeFeedProcessorBuilder()\r\n        .hostName(\"SampleHost_1\")\r\n        .feedContainer(feedContainer)\r\n        .leaseContainer(leaseContainer)\r\n        .options(options)\r\n        .handleChanges(handleChanges())\r\n        .buildChangeFeedProcessor();\r\nchangeFeedProcessorInstance.start()\r\n        .subscribeOn(Schedulers.boundedElastic())\r\n        .subscribe();<\/code><\/pre>\n
    <\/pre>\n
    <\/h5>\n
    Session token mismatch\u00a0 – further optimization<\/h5>\n
    In the previous period we introduced a session token mismatch optimisation<\/a> which allows application developers to configure hints through a\u00a0SessionRetryOptions<\/span><\/span>\u00a0<\/span><\/code>instance. This signals to the SDK whether to pin retries on the local region or move quicker to a remote region, especially when READ_SESSION_NOT_AVAILABLE<\/span><\/code> errors are thrown. The latest change<\/a> adds a minimum time spent in trying to see whether the local time can meet session consistency before attempting to call the remote region. This will add some time the local region can use to catch-up on replication lag.<\/p>\n
    int minMaxRetriesInLocalRegion= 5;\r\nSystem.setProperty(\"COSMOS.MIN_MAX_RETRIES_IN_LOCAL_REGION_WHEN_REMOTE_REGION_PREFERRED\", String.valueOf(minMaxRetriesInLocalRegion));<\/code><\/pre>\n
    <\/h5>\n
    <\/h5>\n
    Change Feed Processor Context for All Versions and Deletes mode in Java SDK<\/h5>\n
    The Change Feed Processor (CFP) has a special mode for All Versions and Deletes<\/a>, giving the user a record of each change to items in the order that it occurred, including intermediate changes to an item between change feed reads, as well as a record of all deletes and the prior image before deletion. This change in the Java SDK<\/a> adds\u00a0ChangeFeedProcessorContext.<\/code>This exposes information on details related to a batch of changes.\u00a0\u00a0<\/bdi><\/p>\n
    public static ChangeFeedProcessor getChangeFeedProcessorForAllVersionsAndDeletesMode(String hostName, CosmosAsyncContainer feedContainer, CosmosAsyncContainer leaseContainer) {\r\n    return new ChangeFeedProcessorBuilder()\r\n            .hostName(hostName)\r\n            .options(options)\r\n            .feedContainer(feedContainer)\r\n            .leaseContainer(leaseContainer)\r\n            .handleAllVersionsAndDeletesChanges((docs, context) -> {\r\n                for (ChangeFeedProcessorItem item : docs) {\r\n                    String leaseToken = context.getLeaseToken();\r\n                    \/\/ Handling of the lease token corresponding to a batch of change feed processor item goes here\r\n                }\r\n            })\r\n            .buildChangeFeedProcessor();\r\n}<\/code><\/pre>\n
    <\/h5>\n
    Hierarchical Partition Key Support in Spark Connector<\/h5>\n
    Users can now use the Spark Connector to create containers with hierarchical partition keys<\/a> in Azure Cosmos DB. In this PySpark sample we create a new container with hierarchical partition keys, ingest some data, then query using the first two levels in the hierarchy:<\/p>\n
    from pyspark.sql.types import StringType\r\nfrom pyspark.sql.functions import udf\r\n\r\ncosmosEndpoint = \"https:\/\/REPLACEME.documents.azure.com:443\/\"\r\ncosmosMasterKey = \"REPLACEME\"\r\n\r\n# Configure Catalog Api to be used\r\nspark.conf.set(\"spark.sql.catalog.cosmosCatalog\", \"com.azure.cosmos.spark.CosmosCatalog\")\r\nspark.conf.set(\"spark.sql.catalog.cosmosCatalog.spark.cosmos.accountEndpoint\", cosmosEndpoint)\r\nspark.conf.set(\"spark.sql.catalog.cosmosCatalog.spark.cosmos.accountKey\", cosmosMasterKey)\r\n\r\n# create an Azure Cosmos DB container with hierarchical partitioning using catalog api\r\ncosmosDatabaseName = \"Database\"\r\ncosmosHierarchicalContainerName = \"HierarchicalPartitionKeyContainer\"\r\nspark.sql(\"CREATE TABLE IF NOT EXISTS cosmosCatalog.{}.{} using cosmos.oltp TBLPROPERTIES(partitionKeyPath = '\/tenantId,\/userId,\/sessionId', manualThroughput = '1100')\".format(cosmosDatabaseName, cosmosHierarchicalContainerName))\r\n\r\ncfg = {\r\n  \"spark.cosmos.accountEndpoint\" : cosmosEndpoint,\r\n  \"spark.cosmos.accountKey\" : cosmosMasterKey,\r\n  \"spark.cosmos.database\" : cosmosDatabaseName,\r\n  \"spark.cosmos.container\" : cosmosHierarchicalContainerName,\r\n  \"spark.cosmos.read.partitioning.strategy\" : \"Restrictive\"\r\n}\r\n\r\n#ingest some data\r\nspark.createDataFrame(((\"id1\", \"tenant 1\", \"User 1\", \"session 1\"), (\"id2\", \"tenant 1\", \"User 1\", \"session 1\"), (\"id3\", \"tenant 2\", \"User 1\", \"session 1\"))) \\\r\n  .toDF(\"id\",\"tenantId\",\"userId\",\"sessionId\") \\\r\n   .write \\\r\n   .format(\"cosmos.oltp\") \\\r\n   .options(**cfg) \\\r\n   .mode(\"APPEND\") \\\r\n   .save()\r\n\r\n#query by filtering the first two levels in the hierarchy without feedRangeFilter - this is less efficient as it will go through all physical partitions\r\nquery_df = spark.read.format(\"cosmos.oltp\").options(**cfg) \\\r\n.option(\"spark.cosmos.read.customQuery\" , \"SELECT * from c where c.tenantId = 'tenant 1' and c.userId = 'User 1'\").load()\r\nquery_df.show()\r\n\r\n# prepare feed range to filter on first two levels in the hierarchy\r\nspark.udf.registerJavaFunction(\"GetFeedRangeForPartitionKey\", \"com.azure.cosmos.spark.udf.GetFeedRangeForHierarchicalPartitionKeyValues\", StringType())\r\npkDefinition = \"{\\\"paths\\\":[\\\"\/tenantId\\\",\\\"\/userId\\\",\\\"\/sessionId\\\"],\\\"kind\\\":\\\"MultiHash\\\"}\"\r\npkValues = \"[\\\"tenant 1\\\", \\\"User 1\\\"]\"\r\nfeedRangeDf = spark.sql(f\"SELECT GetFeedRangeForPartitionKey('{pkDefinition}', '{pkValues}')\")\r\nfeedRange = feedRangeDf.collect()[0][0]\r\n\r\n# query by filtering the first two levels in the hierarchy using feedRangeFilter (will target the physical partition in which all sub-partitions are co-located)\r\nquery_df = spark.read.format(\"cosmos.oltp\").options(**cfg).option(\"spark.cosmos.partitioning.feedRangeFilter\",feedRange).load()\r\nquery_df.show()<\/code><\/pre>\n
    <\/h5>\n
    <\/h5>\n
    Fixes, patches, and enhancements<\/h5>\n
    In addition to all of the above features, we have of course made a large number of smaller bug fixes, security patches, enhancements, and improvements. You can track all the changes for each client library, along with the minimum version we recommend you use<\/strong>, by viewing the change logs:<\/p>\n