changes after further review

Author: Theo van Kraay

src/main/java/com/azure/cosmos/examples/containercrud/async/ContainerCRUDQuickstartAsync.java

@@ -60,9 +60,9 @@ public static void main(String[] args) {
 
     private void containerCRUDDemo() throws Exception {
 
-        logger.info("Using Azure Cosmos DB endpoint: " + AccountSettings.HOST);
+        logger.info("Using Azure Cosmos DB endpoint: {}", AccountSettings.HOST);
 
-        //  Create sync client
+        //  Create async client
         client = new CosmosClientBuilder()
                 .endpoint(AccountSettings.HOST)
                 .key(AccountSettings.MASTER_KEY)
@@ -83,7 +83,7 @@ private void containerCRUDDemo() throws Exception {
 
     // Database Create
     private void createDatabaseIfNotExists() throws Exception {
-        logger.info("Create database " + databaseName + " if not exists...");
+        logger.info("Create database {} if not exists...", databaseName);
 
         //  Create database if not exists
         CosmosDatabaseResponse databaseResponse = client.createDatabaseIfNotExists(databaseName).block();
@@ -94,7 +94,7 @@ private void createDatabaseIfNotExists() throws Exception {
 
     // Container create
     private void createContainerIfNotExists() throws Exception {
-        logger.info("Create container " + containerName + " if not exists.");
+        logger.info("Create container {} if not exists.", containerName);
 
         //  Create container if not exists
         CosmosContainerProperties containerProperties =
@@ -112,7 +112,7 @@ private void createContainerIfNotExists() throws Exception {
 
     // Update container throughput
     private void updateContainerThroughput() throws Exception {
-        logger.info("Update throughput for container " + containerName + ".");
+        logger.info("Update throughput for container {}.", containerName);
 
         // Specify new throughput value
         ThroughputProperties throughputProperties = ThroughputProperties.createManualThroughput(800);
@@ -123,7 +123,7 @@ private void updateContainerThroughput() throws Exception {
 
     // Container read
     private void readContainerById() throws Exception {
-        logger.info("Read container " + containerName + " by ID.");
+        logger.info("Read container {} by ID.", containerName);
 
         //  Read container by ID
         container = database.getContainer(containerName);
@@ -133,7 +133,7 @@ private void readContainerById() throws Exception {
 
     // Container read all
     private void readAllContainers() throws Exception {
-        logger.info("Read all containers in database " + databaseName + ".");
+        logger.info("Read all containers in database {}.", databaseName);
 
         //  Read all containers in the account
         CosmosPagedFlux<CosmosContainerProperties> containers = database.readAllContainers();
@@ -146,7 +146,7 @@ private void readAllContainers() throws Exception {
             + " with request charge of " + readAllContainersResponse.getRequestCharge());
 
             for (CosmosContainerProperties response : readAllContainersResponse.getResults()) {
-                logger.info("container id: "+response.getId());
+                logger.info("container id: {}", response.getId());
                 //Got a page of query result with
             }
             return Flux.empty();
@@ -161,7 +161,7 @@ private void readAllContainers() throws Exception {
 
     // Container delete
     private void deleteAContainer() throws Exception {
-        logger.info("Delete container " + containerName + " by ID.");
+        logger.info("Delete container {} by ID.", containerName);
 
         // Delete container
         CosmosContainerResponse containerResp = database.getContainer(containerName).delete(new CosmosContainerRequestOptions()).block();
@@ -172,7 +172,7 @@ private void deleteAContainer() throws Exception {
 
     // Database delete
     private void deleteADatabase() throws Exception {
-        logger.info("Last step: delete database " + databaseName + " by ID.");
+        logger.info("Last step: delete database {} by ID.", databaseName);
 
         // Delete database
         CosmosDatabaseResponse dbResp = client.getDatabase(databaseName).delete(new CosmosDatabaseRequestOptions()).block();
@@ -184,7 +184,7 @@ private void deleteADatabase() throws Exception {
     // Multi-master only: Container create: last-writer-wins conflict resolution
     // Favor the newest write based on default (internal) timestamp
     private void createContainerIfNotExistsLWWDefaultTimestamp() throws Exception {
-        logger.info("Create container " + containerName + " if not exists.");
+        logger.info("Create container {} if not exists.", containerName);
 
         //  Create container properties structure
         CosmosContainerProperties containerProperties =
@@ -233,7 +233,7 @@ private void createContainerIfNotExistsLWWCustomTimestamp() throws Exception {
     // You will need to upload this stored procedure to the back-end using either
     // the stored procedure sample in this repo, or the Azure portal.
     private void createContainerIfNotExistsCustomSproc() throws Exception {
-        logger.info("Create container " + containerName + " if not exists.");
+        logger.info("Create container {} if not exists.", containerName);
 
         //  Create container properties structure
         CosmosContainerProperties containerProperties =
@@ -259,7 +259,7 @@ private void createContainerIfNotExistsCustomSproc() throws Exception {
     // This method will register the container as using conflict feed;
     // a separate method is needed to actually monitor the conflict feed.
     private void createContainerIfNotExistsCustomConflictFeed() throws Exception {
-        logger.info("Create container " + containerName + " if not exists.");
+        logger.info("Create container {} if not exists.", containerName);
 
         //  Create container properties structure
         CosmosContainerProperties containerProperties =

src/main/java/com/azure/cosmos/examples/databasecrud/async/DatabaseCRUDQuickstartAsync.java

@@ -61,7 +61,7 @@ private void databaseCRUDDemo() throws Exception {
 
         logger.info("Using Azure Cosmos DB endpoint: " + AccountSettings.HOST);
 
-        //  Create sync client
+        //  Create async client
         client = new CosmosClientBuilder()
                 .endpoint(AccountSettings.HOST)
                 .key(AccountSettings.MASTER_KEY)
@@ -79,7 +79,7 @@ private void databaseCRUDDemo() throws Exception {
 
     // Database Create
     private void createDatabaseIfNotExists() throws Exception {
-        logger.info("Create database " + databaseName + " if not exists...");
+        logger.info("Create database {} if not exists...", databaseName);
 
         //  Create database if not exists
         CosmosDatabaseResponse databaseResponse = client.createDatabaseIfNotExists(databaseName).block();
@@ -90,7 +90,7 @@ private void createDatabaseIfNotExists() throws Exception {
 
     // Database read
     private void readDatabaseById() throws Exception {
-        logger.info("Read database " + databaseName + " by ID.");
+        logger.info("Read database {} by ID.", databaseName);
 
         //  Read database by ID
         database = client.getDatabase(databaseName);
@@ -109,9 +109,7 @@ private void readAllDatabases() throws Exception {
         String msg="Listing databases in account:\n";
 
         databases.byPage(100).flatMap(readAllDatabasesResponse -> {
-            logger.info("read " +
-            readAllDatabasesResponse.getResults().size() + " database(s)"
-            + " with request charge of " + readAllDatabasesResponse.getRequestCharge());
+            logger.info("read {} database(s) with request charge of {}", readAllDatabasesResponse.getResults().size(),readAllDatabasesResponse.getRequestCharge());
 
             for (CosmosDatabaseProperties response : readAllDatabasesResponse.getResults()) {
                 logger.info("database id: "+response.getId());
@@ -127,7 +125,7 @@ private void readAllDatabases() throws Exception {
 
     // Database delete
     private void deleteADatabase() throws Exception {
-        logger.info("Last step: delete database " + databaseName + " by ID.");
+        logger.info("Last step: delete database {} by ID.", databaseName);
 
         // Delete database
         CosmosDatabaseResponse dbResp = client.getDatabase(databaseName).delete(new CosmosDatabaseRequestOptions()).block();

src/main/java/com/azure/cosmos/examples/documentcrud/async/DocumentCRUDQuickstartAsync.java

@@ -53,6 +53,8 @@ public class DocumentCRUDQuickstartAsync {
     Family family = new Family();
     Family family2 = new Family();
 
+    List<Family> families = new ArrayList<>();
+
     protected static Logger logger = LoggerFactory.getLogger(DocumentCRUDQuickstartAsync.class);
 
     public void close() {
@@ -79,7 +81,7 @@ public static void main(String[] args) {
             logger.info("Demo complete, please hold while resources are released");
         } catch (Exception e) {
             e.printStackTrace();
-            logger.error(String.format("Cosmos getStarted failed with %s", e));
+            logger.error("Cosmos getStarted failed with {}", e);
         } finally {
             logger.info("Closing the client");
             p.shutdown();
@@ -103,7 +105,7 @@ private void documentCRUDDemo() throws Exception {
         createContainerIfNotExists();
 
         //add docs to array
-        List<Family> families = new ArrayList<>();
+        
         families.add(Families.getAndersenFamilyItem());
         families.add(Families.getWakefieldFamilyItem());
         families.add(Families.getJohnsonFamilyItem());
@@ -210,7 +212,7 @@ private void createDocument() throws Exception {
     }
 
     private void createDocuments(Flux<Family> families) throws Exception {
-        logger.info("Create documents {}", documentId);
+        logger.info("Create documents {}", this.families.stream().map(x -> x.getId()).collect(Collectors.toList()));
 
         // Define a document as a POJO (internally this
         // is converted to JSON via custom serialization)
@@ -220,10 +222,8 @@ private void createDocuments(Flux<Family> families) throws Exception {
                     return container.createItem(family);
                 }) // Flux of item request responses
                 .flatMap(itemResponse -> {
-                    logger.info(String.format("Created item with request charge of %.2f within" +
-                                    " duration %s",
-                            itemResponse.getRequestCharge(), itemResponse.getDuration()));
-                    logger.info(String.format("Item ID: %s\n", itemResponse.getItem().getId()));
+                    logger.info("Created item with request charge of {} within duration {}", itemResponse.getRequestCharge(), itemResponse.getDuration());
+                    logger.info("Item ID: {}\n", itemResponse.getItem().getId());
                     return Mono.just(itemResponse.getRequestCharge());
                 }) // Flux of request charges
                 .reduce(0.0,
@@ -232,7 +232,7 @@ private void createDocuments(Flux<Family> families) throws Exception {
                 .doOnSuccess(itemResponse -> {
                     logger.info("Aggregated charge (all decimals): {}", itemResponse);
                     // Preserve the total charge and print aggregate charge/item count stats.
-                    logger.info(String.format("Created items with total request charge of %.2f\n", itemResponse));
+                    logger.info("Created items with total request charge of {}\n", itemResponse);
                 })
                 .doOnError((exception) -> {
                     logger.error(
@@ -366,45 +366,33 @@ private void upsertDocument() throws Exception {
     private void replaceDocumentWithConditionalEtagCheck() throws Exception {
         logger.info("Replace document {}, employing optimistic concurrency using ETag.", documentId);
 
-        // Obtained current document ETag
+        //chain reading and replacing of item together in reactive stream
         container.readItem(documentId, new PartitionKey(documentLastName), Family.class)
-                .doOnSuccess(itemResponse -> {
-                    logger.info("Request charge of readItem operation: {} RU", itemResponse.getRequestCharge());
-                    this.family = itemResponse.getItem();
-                    this.etag1 = itemResponse.getETag();
-                })
-                .doOnError((exception) -> {
-                    logger.error(
-                            "Exception 1. e: {}",
-                            exception.getLocalizedMessage(),
-                            exception);
-                }).subscribe();
-
-        // We are adding Thread.sleep to mimic the some business computation that can
-        // happen while waiting for earlier processes to finish.
-        Thread.sleep(1000);
-
-        String etag = this.etag1;
-        logger.info("Read document {} to obtain current ETag: {}", documentId, etag);
-
-        // Modify document
-        Family family = this.family;
-        family.setRegistered(!family.isRegistered());
-
-        // Persist the change back to the server, updating the ETag in the process
-        // This models a concurrent change made to the document
-        container.replaceItem(family, family.getId(), new PartitionKey(family.getLastName()), new CosmosItemRequestOptions())
-                .doOnSuccess(itemResponse -> {
-                    logger.info("'Concurrent' update to document {} so ETag is now {}",documentId, itemResponse.getResponseHeaders().get("etag"));
-                })
-                .doOnError((exception) -> {
-                    logger.error(
-                            "Exception 2. e: {}",
-                            exception.getLocalizedMessage(),
-                            exception);
+                .flatMap(response -> { 
+                    logger.info("Read document {} to obtain current ETag: {}", documentId, response.getETag());
+                    //set instance variables for family nd etag1 to use later
+                    this.family = response.getItem();
+                    this.etag1 = response.getETag();
+                    logger.info("Request charge of readItem operation: {} RU", response.getRequestCharge());       
+                    Family family = this.family;
+                    family.setRegistered(!family.isRegistered());
+                    // Persist the change back to the server, updating the ETag in the process
+                    // This models a concurrent change made to the document
+                    return container
+                            .replaceItem(family, family.getId(), new PartitionKey(family.getLastName()),
+                                    new CosmosItemRequestOptions())
+                            .doOnSuccess(itemResponse -> {
+                                logger.info("'Concurrent' update to document {} so ETag is now {}", documentId,
+                                        itemResponse.getResponseHeaders().get("etag"));
+                            })
+                            .doOnError((exception) -> {
+                                logger.error(
+                                        "Exception e: {}",
+                                        exception.getLocalizedMessage(),
+                                        exception);
+                            });
                 }).subscribe();
 
-
         // We are adding Thread.sleep to mimic the some business computation that can
         // happen while waiting for earlier processes to finish.
         Thread.sleep(1000);
@@ -413,7 +401,8 @@ private void replaceDocumentWithConditionalEtagCheck() throws Exception {
         // This should fail because the "concurrent" document change updated the ETag.
 
         CosmosItemRequestOptions requestOptions = new CosmosItemRequestOptions();
-        requestOptions.setIfMatchETag(etag);
+        //using etag1 instance variable saved from earlier update to etag
+        requestOptions.setIfMatchETag(this.etag1.toString());
 
         family.setDistrict("Seafood");
 
@@ -454,39 +443,32 @@ private void readDocumentOnlyIfChanged() throws Exception {
         String etag = this.etag2;
 
         // Re-read doc but with conditional access requirement that ETag has changed.
-        // This should fail.
+        // This should fail (304).
         CosmosItemRequestOptions requestOptions = new CosmosItemRequestOptions();
         requestOptions.setIfNoneMatchETag(etag);
 
-        container
-                .readItem(documentId, new PartitionKey(documentLastName), requestOptions, Family.class)
-                .doOnSuccess(itemResponse -> {
-                    logger.info("Re-read doc with status code of {} (we anticipate failure due to ETag not having changed.)", itemResponse.getStatusCode());
-                })
-                .doOnError((exception) -> {
-                    logger.info("Exception. e: {}");
-                })
-                .onErrorResume(exception -> Mono.empty())
-                .subscribe();
-
-        // We are adding Thread.sleep to mimic the some business computation that can
-        // happen while waiting for earlier processes to finish.
-        Thread.sleep(1000);
-
-        // Replace the doc with a modified version, which will update ETag
-        Family family = this.family2;
-        family.setRegistered(!family.isRegistered());
-        container.replaceItem(family, family.getId(), new PartitionKey(family.getLastName()), new CosmosItemRequestOptions())
-                .doOnSuccess(itemResponse -> {
-                    logger.info("Request charge of replace operation: {} RU", itemResponse.getRequestCharge());
-                    logger.info("Modified and replaced the doc (updates ETag.)");
-                })
-                .doOnError((exception) -> {
-                    logger.error(
-                            "Exception. e: {}",
-                            exception.getLocalizedMessage(),
-                            exception);
-                }).subscribe();
+        //chain reading and replacing of item together in reactive stream
+        container.readItem(documentId, new PartitionKey(documentLastName), requestOptions, Family.class)
+        .flatMap(response -> { 
+            logger.info("Re-read doc with status code of {} (we anticipate 304 failure due to ETag not having changed.)", response.getStatusCode());
+            logger.info("Request charge of readItem operation: {} RU", response.getRequestCharge());       
+            Family family = this.family2;
+            family.setRegistered(!family.isRegistered());
+            // Replace the doc with a modified version, which will update ETag
+            return container
+                    .replaceItem(family, family.getId(), new PartitionKey(family.getLastName()),
+                            new CosmosItemRequestOptions())
+                    .doOnSuccess(itemResponse -> {
+                        logger.info("'Concurrent' update to document {} so ETag is now {}", documentId,
+                                itemResponse.getResponseHeaders().get("etag"));
+                    })
+                    .doOnError((exception) -> {
+                        logger.error(
+                                "Exception e: {}",
+                                exception.getLocalizedMessage(),
+                                exception);
+                    });
+        }).subscribe();
 
         // We are adding Thread.sleep to mimic the some business computation that can
         // happen while waiting for earlier processes to finish.

src/main/java/com/azure/cosmos/examples/queries/async/QueriesQuickstartAsync.java

@@ -95,7 +95,7 @@ private void queriesDemo() throws Exception {
 
         logger.info("Using Azure Cosmos DB endpoint: {}", AccountSettings.HOST);
 
-        //  Create sync client
+        //  Create async client
         client = new CosmosClientBuilder()
                 .endpoint(AccountSettings.HOST)
                 .key(AccountSettings.MASTER_KEY)
@@ -459,7 +459,7 @@ private void deleteADocument() throws Exception {
         logger.info("Delete document {} by ID.", documentId);
 
         // Delete document
-        container.deleteItem(documentId, new PartitionKey(documentLastName), new CosmosItemRequestOptions());
+        container.deleteItem(documentId, new PartitionKey(documentLastName), new CosmosItemRequestOptions()).block();
 
         logger.info("Done.");
     }