Cosmos: CFP 3.3.2 hot-fixes

sdk/cosmos/README.md

@@ -92,7 +92,7 @@ For example, using maven, you can add the following dependency to your maven pom
 <dependency>
   <groupId>com.microsoft.azure</groupId>
   <artifactId>azure-cosmos</artifactId>
-  <version>3.3.0</version>
+  <version>3.3.2</version>
 </dependency>
 ```
 [//]: # ({x-version-update-end})

sdk/cosmos/changelog/README.md

@@ -1,5 +1,11 @@
 ## Changelog
 
+### 3.3.2
+- ChangeFeedProcessor; fixes and extra logging related to the creations of the lease documents.
+- Port consistency policy bug fix (see https://github.com/Azure/azure-cosmosdb-java/pull/196)
+- Port test fixes (see https://github.com/Azure/azure-cosmosdb-java/pull/196)
+- ChangeFeedProcessor: fix a race condition issue where the lease document has been updated while we are trying to write the current state like when check-pointing the continuation token.
+
 ### 3.3.1
 - Added @JsonIgnore on getLogger in JsonSerializable
 

sdk/cosmos/microsoft-azure-cosmos-benchmark/pom.xml

@@ -8,12 +8,12 @@ Licensed under the MIT License.
   <parent>
     <groupId>com.microsoft.azure</groupId>
     <artifactId>azure-cosmos-parent</artifactId>
-    <version>3.3.1</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos-parent;current} -->
+    <version>3.3.2</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos-parent;current} -->
   </parent>
 
   <groupId>com.microsoft.azure</groupId>
   <artifactId>azure-cosmos-benchmark</artifactId>
-  <version>3.3.1</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos-benchmark;current} -->
+  <version>3.3.2</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos-benchmark;current} -->
   <name>Microsoft Azure SDK for SQL API of Azure Cosmos DB Service - Benchmarking tool</name>
   <description>This package contains Benchmarking tool for Microsoft Azure SDK for SQL API of Azure Cosmos DB Service</description>
   <url>https://github.com/Azure/azure-sdk-for-java</url>
@@ -33,7 +33,7 @@ Licensed under the MIT License.
     <dependency>
       <groupId>com.microsoft.azure</groupId>
       <artifactId>azure-cosmos</artifactId>
-      <version>3.3.1</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos;current} -->
+      <version>3.3.2</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos;current} -->
     </dependency>
 
     <dependency>

sdk/cosmos/microsoft-azure-cosmos-examples/pom.xml

@@ -8,12 +8,12 @@ Licensed under the MIT License.
   <parent>
     <groupId>com.microsoft.azure</groupId>
     <artifactId>azure-cosmos-parent</artifactId>
-    <version>3.3.1</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos-parent;current} -->
+    <version>3.3.2</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos-parent;current} -->
   </parent>
 
   <groupId>com.microsoft.azure</groupId>
   <artifactId>azure-cosmos-examples</artifactId>
-  <version>3.3.1</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos-examples;current} -->
+  <version>3.3.2</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos-examples;current} -->
   <name>Microsoft Azure SDK for SQL API of Azure Cosmos DB Service - Examples</name>
   <description>This package contains examples for Microsoft Azure SDK for SQL API of Azure Cosmos DB Service</description>
   <url>https://github.com/Azure/azure-sdk-for-java</url>
@@ -89,7 +89,7 @@ Licensed under the MIT License.
     <dependency>
       <groupId>com.microsoft.azure</groupId>
       <artifactId>azure-cosmos</artifactId>
-      <version>3.3.1</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos;current} -->
+      <version>3.3.2</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos;current} -->
     </dependency>
 
     <dependency>

sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/ConflictAPITest.java

@@ -15,6 +15,7 @@
 import com.azure.data.cosmos.FeedResponse;
 import com.azure.data.cosmos.PartitionKeyDefinition;
 import com.azure.data.cosmos.internal.HttpConstants;
+import com.azure.data.cosmos.internal.ResourceResponse;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Ignore;
@@ -81,10 +82,12 @@ public void setUp() {
 
         int numberOfDocuments = 20;
         // Add documents
+        List<Mono<Void>> tasks = new ArrayList<>();
         for (int i = 0; i < numberOfDocuments; i++) {
             Document doc = new Document(String.format("{ 'id': 'loc%d', 'counter': %d}", i, i));
-            client.createDocument(getCollectionLink(), doc, null, true).single().block();
+            tasks.add(client.createDocument(getCollectionLink(), doc, null, true).then());
         }
+        Flux.merge(tasks).then().block();
     }
 
     @AfterClass(groups = "samples", timeOut = TIMEOUT)

sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/DocumentQueryAsyncAPITest.java

@@ -106,10 +106,12 @@ public void setUp() {
 
         numberOfDocuments = 20;
         // Add documents
+        List<Mono<Void>> tasks = new ArrayList<>();
         for (int i = 0; i < numberOfDocuments; i++) {
             Document doc = new Document(String.format("{ 'id': 'loc%d', 'counter': %d}", i, i));
-            client.createDocument(getCollectionLink(), doc, null, true).single().block();
+            tasks.add(client.createDocument(getCollectionLink(), doc, null, true).then());
         }
+        Flux.merge(tasks).then().block();
     }
 
     @AfterClass(groups = "samples", timeOut = TIMEOUT)

sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/InMemoryGroupbyTest.java

@@ -21,6 +21,7 @@
 import org.testng.annotations.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.GroupedFlux;
+import reactor.core.publisher.Mono;
 
 import java.time.LocalDateTime;
 import java.util.ArrayList;
@@ -68,6 +69,7 @@ public void setUp() throws Exception {
 
         int numberOfPayers = 10;
         int numberOfDocumentsPerPayer = 10;
+        List<Mono<Void>> tasks = new ArrayList<>();
 
         for (int i = 0; i < numberOfPayers; i++) {
 
@@ -81,11 +83,10 @@ public void setUp() throws Exception {
                         + "'payer_id': %d, "
                         + " 'created_time' : %d "
                         + "}", UUID.randomUUID().toString(), i, currentTime.getSecond()));
-                client.createDocument(getCollectionLink(), doc, null, true).single().block();
-
-                Thread.sleep(100);
+                tasks.add(client.createDocument(getCollectionLink(), doc, null, true).then());
             }
         }
+        Flux.merge(tasks).then().block();
         System.out.println("finished inserting documents");
     }
 

sdk/cosmos/microsoft-azure-cosmos/pom.xml

@@ -7,12 +7,12 @@ Licensed under the MIT License.
   <parent>
     <groupId>com.microsoft.azure</groupId>
     <artifactId>azure-cosmos-parent</artifactId>
-    <version>3.3.1</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos-parent;current} -->
+    <version>3.3.2</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos-parent;current} -->
   </parent>
 
   <groupId>com.microsoft.azure</groupId>
   <artifactId>azure-cosmos</artifactId>
-  <version>3.3.1</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos;current} -->
+  <version>3.3.2</version> <!-- {x-version-update;com.microsoft.azure:azure-cosmos;current} -->
   <name>Microsoft Azure SDK for SQL API of Azure Cosmos DB Service</name>
   <description>This Package contains Microsoft Azure Cosmos SDK (with Reactive Extension rx support) for Azure Cosmos DB SQL API</description>
   <packaging>jar</packaging>

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/HttpConstants.java

@@ -252,7 +252,7 @@ public static class Versions {
         // TODO: FIXME we can use maven plugin for generating a version file
         // @see
         // https://stackoverflow.com/questions/2469922/generate-a-version-java-file-in-maven
-        public static final String SDK_VERSION = "3.3.1";
+        public static final String SDK_VERSION = "3.3.2";
         public static final String SDK_NAME = "cosmosdb-java-sdk";
     }
 

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/ChangeFeedObserverContext.java

@@ -34,5 +34,5 @@ public interface ChangeFeedObserverContext {
      *
      * @return a representation of the deferred computation of this call.
      */
-    Mono<Void> checkpoint();
+    Mono<Lease> checkpoint();
 }

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/PartitionCheckpointer.java

@@ -14,5 +14,5 @@ public interface PartitionCheckpointer {
      * @param сontinuationToken the continuation token.
      * @return a deferred operation of this call.
      */
-    Mono<Void> checkpointPartition(String сontinuationToken);
+    Mono<Lease> checkpointPartition(String сontinuationToken);
 }

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/exceptions/LeaseConflictException.java

@@ -0,0 +1,69 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.data.cosmos.internal.changefeed.exceptions;
+
+import com.azure.data.cosmos.internal.changefeed.Lease;
+
+/**
+ * Exception occurred when the lease was updated by a different thread or worker while current thread is trying to update it as well.
+ */
+public class LeaseConflictException extends RuntimeException {
+    private static final String DEFAULT_MESSAGE = "The lease was updated by a different worker.";
+    private Lease lease;
+
+    /**
+     * Initializes a new instance of the @link LeaseConflictException} class.
+     */
+    public LeaseConflictException() {
+    }
+
+    /**
+     * Initializes a new instance of the @link LeaseConflictException} class using the specified lease.
+     *
+     * @param lease an instance of a lost lease.
+     */
+    public LeaseConflictException(Lease lease) {
+        super(DEFAULT_MESSAGE);
+        this.lease = lease;
+    }
+
+    /**
+     * Initializes a new instance of the @link LeaseConflictException} class using the specified lease.
+     *
+     * @param lease an instance of a lost lease.
+     * @param message the exception error message.
+     */
+    public LeaseConflictException(Lease lease, String message) {
+        super(message);
+        this.lease = lease;
+    }
+
+    /**
+     * Initializes a new instance of the @link LeaseConflictException} class using error message.
+     *
+     * @param message the exception error message.
+     */
+    public LeaseConflictException(String message) {
+        super(message);
+    }
+
+    /**
+     * Initializes a new instance of the @link LeaseConflictException} class using error message and inner exception.
+     *
+     * @param message the exception error message.
+     * @param innerException the inner exception.
+     *
+     */
+    public LeaseConflictException(String message, Exception innerException) {
+        super(message, innerException.getCause());
+    }
+
+    /**
+     * Gets the lost lease.
+     *
+     * @return the lost lease.
+     */
+    public Lease getLease() {
+        return this.lease;
+    }
+}

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/exceptions/LeaseLostException.java

@@ -17,17 +17,15 @@ public class LeaseLostException extends RuntimeException {
     /**
      * Initializes a new instance of the @link LeaseLostException} class.
      */
-    public LeaseLostException()
-    {
+    public LeaseLostException() {
     }
 
     /**
      * Initializes a new instance of the @link LeaseLostException} class using the specified lease.
      *
      * @param lease an instance of a lost lease.
      */
-    public LeaseLostException(Lease lease)
-    {
+    public LeaseLostException(Lease lease) {
         super(DEFAULT_MESSAGE);
         this.lease = lease;
     }
@@ -37,8 +35,7 @@ public LeaseLostException(Lease lease)
      *
      * @param message the exception error message.
      */
-    public LeaseLostException(String message)
-    {
+    public LeaseLostException(String message) {
         super(message);
     }
 
@@ -49,8 +46,7 @@ public LeaseLostException(String message)
      * @param innerException the inner exception.
      *
      */
-    public LeaseLostException(String message, Exception innerException)
-    {
+    public LeaseLostException(String message, Exception innerException) {
         super(message, innerException.getCause());
     }
 
@@ -62,8 +58,7 @@ public LeaseLostException(String message, Exception innerException)
      * @param innerException the inner exception.
      * @param isGone true if lease doesn't exist.
      */
-    public LeaseLostException(Lease lease, Exception innerException, boolean isGone)
-    {
+    public LeaseLostException(Lease lease, Exception innerException, boolean isGone) {
         super(DEFAULT_MESSAGE, innerException.getCause());
         this.lease = lease;
         this.isGone = isGone;

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/AutoCheckpointer.java

@@ -7,6 +7,8 @@
 import com.azure.data.cosmos.internal.changefeed.ChangeFeedObserverCloseReason;
 import com.azure.data.cosmos.internal.changefeed.ChangeFeedObserverContext;
 import com.azure.data.cosmos.internal.changefeed.CheckpointFrequency;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
@@ -18,6 +20,7 @@
  * Auto check-pointer implementation for {@link ChangeFeedObserver}.
  */
 class AutoCheckpointer implements ChangeFeedObserver {
+    private final Logger logger = LoggerFactory.getLogger(AutoCheckpointer.class);
     private final CheckpointFrequency checkpointFrequency;
     private final ChangeFeedObserver observer;
     private volatile int processedDocCount;
@@ -50,6 +53,9 @@ public void close(ChangeFeedObserverContext context, ChangeFeedObserverCloseReas
     @Override
     public Mono<Void> processChanges(ChangeFeedObserverContext context, List<CosmosItemProperties> docs) {
         return this.observer.processChanges(context, docs)
+            .doOnError(throwable -> {
+                logger.warn("Unexpected exception from thread {}", Thread.currentThread().getId(), throwable);
+            })
             .then(this.afterProcessChanges(context));
     }
 
@@ -58,10 +64,14 @@ private Mono<Void> afterProcessChanges(ChangeFeedObserverContext context) {
 
         if (this.isCheckpointNeeded()) {
             return context.checkpoint()
-                .doOnSuccess((Void) -> {
+                .doOnError(throwable -> {
+                    logger.warn("Checkpoint failed; this worker will be killed", throwable);
+                })
+                .doOnSuccess( (Void) -> {
                     this.processedDocCount = 0;
                     this.lastCheckpointTime = ZonedDateTime.now(ZoneId.of("UTC"));
-                });
+                })
+                .then();
         }
         return Mono.empty();
     }

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/BootstrapperImpl.java

@@ -61,6 +61,7 @@ public Mono<Void> initialize() {
                 if (initialized) {
                     return Mono.empty();
                 } else {
+                    logger.info("Acquire initialization lock");
                     return this.leaseStore.acquireInitializationLock(this.lockTime)
                         .flatMap(lockAcquired -> {
                             this.isLockAcquired = lockAcquired;
@@ -74,7 +75,7 @@ public Mono<Void> initialize() {
                             }
                         })
                         .onErrorResume(throwable -> {
-                            logger.warn("Unexpected exception caught", throwable);
+                            logger.warn("Unexpected exception caught while initializing the lock", throwable);
                             return Mono.just(this.isLockAcquired);
                         })
                         .flatMap(lockAcquired -> {

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/ChangeFeedObserverContextImpl.java

@@ -6,6 +6,7 @@
 import com.azure.data.cosmos.CosmosItemProperties;
 import com.azure.data.cosmos.FeedResponse;
 import com.azure.data.cosmos.internal.changefeed.ChangeFeedObserverContext;
+import com.azure.data.cosmos.internal.changefeed.Lease;
 import com.azure.data.cosmos.internal.changefeed.PartitionCheckpointer;
 import reactor.core.publisher.Mono;
 
@@ -40,7 +41,7 @@ public ChangeFeedObserverContextImpl(String leaseToken, FeedResponse<CosmosItemP
      * @return a deferred computation of this call.
      */
     @Override
-    public Mono<Void> checkpoint() {
+    public Mono<Lease> checkpoint() {
         this.responseContinuation = this.feedResponse.continuationToken();
 
         return this.checkpointer.checkpointPartition(this.responseContinuation);

sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DefaultObserver.java

@@ -34,8 +34,13 @@ public void close(ChangeFeedObserverContext context, ChangeFeedObserverCloseReas
     @Override
     public Mono<Void> processChanges(ChangeFeedObserverContext context, List<CosmosItemProperties> docs) {
         log.info("Start processing from thread {}", Thread.currentThread().getId());
-        consumer.accept(docs);
-        log.info("Done processing from thread {}", Thread.currentThread().getId());
+        try {
+            consumer.accept(docs);
+            log.info("Done processing from thread {}", Thread.currentThread().getId());
+        } catch (Exception ex) {
+            log.warn("Unexpected exception thrown from thread {}", Thread.currentThread().getId(), ex);
+            return Mono.error(ex);
+        }
 
         return Mono.empty();
     }