From 268f6f0265db17c7368c15a43104f5dec251218b Mon Sep 17 00:00:00 2001 From: milismsft-mac Date: Thu, 7 Nov 2019 10:24:14 -0800 Subject: [PATCH 1/2] ChangeFeedProcessor; fixes and extra logging related to the creations of the lease documents. Port consistency policy bug fix from V2 (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. --- sdk/cosmos/README.md | 2 +- sdk/cosmos/changelog/README.md | 6 ++ .../microsoft-azure-cosmos-benchmark/pom.xml | 6 +- .../microsoft-azure-cosmos-examples/pom.xml | 6 +- .../cosmos/rx/examples/ConflictAPITest.java | 5 +- .../examples/DocumentQueryAsyncAPITest.java | 4 +- .../rx/examples/InMemoryGroupbyTest.java | 7 +- sdk/cosmos/microsoft-azure-cosmos/pom.xml | 4 +- .../changefeed/ChangeFeedObserverContext.java | 2 +- .../changefeed/PartitionCheckpointer.java | 2 +- .../exceptions/LeaseConflictException.java | 69 +++++++++++++++++++ .../exceptions/LeaseLostException.java | 15 ++-- .../implementation/AutoCheckpointer.java | 14 +++- .../implementation/BootstrapperImpl.java | 3 +- .../ChangeFeedObserverContextImpl.java | 3 +- .../implementation/DefaultObserver.java | 9 ++- .../DocumentServiceLeaseStore.java | 20 ++++-- .../DocumentServiceLeaseUpdaterImpl.java | 27 ++++++-- .../implementation/LeaseRenewerImpl.java | 9 ++- .../implementation/LeaseStoreManagerImpl.java | 8 ++- ...onWrappingChangeFeedObserverDecorator.java | 9 ++- .../PartitionCheckpointerImpl.java | 5 +- .../PartitionProcessorImpl.java | 36 +++++++--- .../directconnectivity/ConsistencyWriter.java | 38 ++++------ .../directconnectivity/StoreReader.java | 12 +++- .../rntbd/RntbdRequestManager.java | 2 +- .../DocumentQuerySpyWireContentTest.java | 22 +++--- .../data/cosmos/internal/SessionTest.java | 14 ++-- .../GatewayAddressCacheTest.java | 5 +- .../internal/query/DocumentProducerTest.java | 2 +- sdk/cosmos/pom.xml | 2 +- 31 files changed, 257 insertions(+), 111 deletions(-) create mode 100644 sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/exceptions/LeaseConflictException.java diff --git a/sdk/cosmos/README.md b/sdk/cosmos/README.md index 338d2e5d7a68..4727992f92e7 100644 --- a/sdk/cosmos/README.md +++ b/sdk/cosmos/README.md @@ -92,7 +92,7 @@ For example, using maven, you can add the following dependency to your maven pom com.microsoft.azure azure-cosmos - 3.3.0 + 3.3.2 ``` [//]: # ({x-version-update-end}) diff --git a/sdk/cosmos/changelog/README.md b/sdk/cosmos/changelog/README.md index 3c00e3cf3287..7edaa44a7b9c 100644 --- a/sdk/cosmos/changelog/README.md +++ b/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 diff --git a/sdk/cosmos/microsoft-azure-cosmos-benchmark/pom.xml b/sdk/cosmos/microsoft-azure-cosmos-benchmark/pom.xml index 7604f27acb91..45410b8a565d 100644 --- a/sdk/cosmos/microsoft-azure-cosmos-benchmark/pom.xml +++ b/sdk/cosmos/microsoft-azure-cosmos-benchmark/pom.xml @@ -8,12 +8,12 @@ Licensed under the MIT License. com.microsoft.azure azure-cosmos-parent - 3.3.1 + 3.3.2 com.microsoft.azure azure-cosmos-benchmark - 3.3.1 + 3.3.2 Microsoft Azure SDK for SQL API of Azure Cosmos DB Service - Benchmarking tool This package contains Benchmarking tool for Microsoft Azure SDK for SQL API of Azure Cosmos DB Service https://github.com/Azure/azure-sdk-for-java @@ -33,7 +33,7 @@ Licensed under the MIT License. com.microsoft.azure azure-cosmos - 3.3.1 + 3.3.2 diff --git a/sdk/cosmos/microsoft-azure-cosmos-examples/pom.xml b/sdk/cosmos/microsoft-azure-cosmos-examples/pom.xml index 11662f8db5bc..b7abff3abccf 100644 --- a/sdk/cosmos/microsoft-azure-cosmos-examples/pom.xml +++ b/sdk/cosmos/microsoft-azure-cosmos-examples/pom.xml @@ -8,12 +8,12 @@ Licensed under the MIT License. com.microsoft.azure azure-cosmos-parent - 3.3.1 + 3.3.2 com.microsoft.azure azure-cosmos-examples - 3.3.1 + 3.3.2 Microsoft Azure SDK for SQL API of Azure Cosmos DB Service - Examples This package contains examples for Microsoft Azure SDK for SQL API of Azure Cosmos DB Service https://github.com/Azure/azure-sdk-for-java @@ -89,7 +89,7 @@ Licensed under the MIT License. com.microsoft.azure azure-cosmos - 3.3.1 + 3.3.2 diff --git a/sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/ConflictAPITest.java b/sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/ConflictAPITest.java index 4901b8aa7f03..0cac91a23212 100644 --- a/sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/ConflictAPITest.java +++ b/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> 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) diff --git a/sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/DocumentQueryAsyncAPITest.java b/sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/DocumentQueryAsyncAPITest.java index 6f478e033a76..7a11b6419b93 100644 --- a/sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/DocumentQueryAsyncAPITest.java +++ b/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> 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) diff --git a/sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/InMemoryGroupbyTest.java b/sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/InMemoryGroupbyTest.java index b93b2cd68f96..f51c5c321d01 100644 --- a/sdk/cosmos/microsoft-azure-cosmos-examples/src/test/java/com/azure/data/cosmos/rx/examples/InMemoryGroupbyTest.java +++ b/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> 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"); } diff --git a/sdk/cosmos/microsoft-azure-cosmos/pom.xml b/sdk/cosmos/microsoft-azure-cosmos/pom.xml index abe32ef363c3..44b185269f64 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/pom.xml +++ b/sdk/cosmos/microsoft-azure-cosmos/pom.xml @@ -7,12 +7,12 @@ Licensed under the MIT License. com.microsoft.azure azure-cosmos-parent - 3.3.1 + 3.3.2 com.microsoft.azure azure-cosmos - 3.3.1 + 3.3.2 Microsoft Azure SDK for SQL API of Azure Cosmos DB Service This Package contains Microsoft Azure Cosmos SDK (with Reactive Extension rx support) for Azure Cosmos DB SQL API jar diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/ChangeFeedObserverContext.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/ChangeFeedObserverContext.java index 1e408ed9eb91..17aac33282fb 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/ChangeFeedObserverContext.java +++ b/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 checkpoint(); + Mono checkpoint(); } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/PartitionCheckpointer.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/PartitionCheckpointer.java index 1517b68f28c9..91db2d5d9379 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/PartitionCheckpointer.java +++ b/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 checkpointPartition(String сontinuationToken); + Mono checkpointPartition(String сontinuationToken); } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/exceptions/LeaseConflictException.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/exceptions/LeaseConflictException.java new file mode 100644 index 000000000000..978972a6cc22 --- /dev/null +++ b/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; + } +} diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/exceptions/LeaseLostException.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/exceptions/LeaseLostException.java index fd58128e3989..bf19d27453aa 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/exceptions/LeaseLostException.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/exceptions/LeaseLostException.java @@ -17,8 +17,7 @@ public class LeaseLostException extends RuntimeException { /** * Initializes a new instance of the @link LeaseLostException} class. */ - public LeaseLostException() - { + public LeaseLostException() { } /** @@ -26,8 +25,7 @@ public LeaseLostException() * * @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; diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/AutoCheckpointer.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/AutoCheckpointer.java index b064834a1c1b..71e33178de11 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/AutoCheckpointer.java +++ b/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 processChanges(ChangeFeedObserverContext context, List 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 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(); } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/BootstrapperImpl.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/BootstrapperImpl.java index ab4b5385d56a..0ce8aebec490 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/BootstrapperImpl.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/BootstrapperImpl.java @@ -61,6 +61,7 @@ public Mono 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 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 -> { diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/ChangeFeedObserverContextImpl.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/ChangeFeedObserverContextImpl.java index d15d45a486c0..218a96f88ded 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/ChangeFeedObserverContextImpl.java +++ b/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 checkpoint() { + public Mono checkpoint() { this.responseContinuation = this.feedResponse.continuationToken(); return this.checkpointer.checkpointPartition(this.responseContinuation); diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DefaultObserver.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DefaultObserver.java index cc64f34c94ca..683b7f26d533 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DefaultObserver.java +++ b/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 processChanges(ChangeFeedObserverContext context, List 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(); } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DocumentServiceLeaseStore.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DocumentServiceLeaseStore.java index 8e29ae6f4112..cd60e4b9af6d 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DocumentServiceLeaseStore.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DocumentServiceLeaseStore.java @@ -14,6 +14,8 @@ import com.azure.data.cosmos.internal.changefeed.LeaseStore; import com.azure.data.cosmos.internal.changefeed.RequestOptionsFactory; import com.azure.data.cosmos.internal.changefeed.ServiceItemLease; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import reactor.core.publisher.Mono; import java.time.Duration; @@ -22,6 +24,7 @@ * Implementation for LeaseStore. */ class DocumentServiceLeaseStore implements LeaseStore { + private final Logger logger = LoggerFactory.getLogger(BootstrapperImpl.class); private ChangeFeedContextClient client; private String containerNamePrefix; private CosmosContainer leaseCollectionLink; @@ -51,16 +54,18 @@ public Mono isInitialized() { CosmosItemRequestOptions requestOptions = this.requestOptionsFactory.createRequestOptions( ServiceItemLease.fromDocument(doc)); - CosmosItem docItem = this.client.getContainerClient().getItem(markerDocId, "/id"); + CosmosItem docItem = this.client.getContainerClient().getItem(markerDocId, markerDocId); return this.client.readItem(docItem, requestOptions) .flatMap(documentResourceResponse -> Mono.just(documentResourceResponse.item() != null)) .onErrorResume(throwable -> { if (throwable instanceof CosmosClientException) { CosmosClientException e = (CosmosClientException) throwable; if (e.statusCode() == ChangeFeedHelper.HTTP_STATUS_CODE_NOT_FOUND) { + logger.info("Lease synchronization document not found"); return Mono.just(false); } } + logger.error("Unexpected exception thrown", throwable); return Mono.error(throwable); }); } @@ -71,15 +76,17 @@ public Mono markInitialized() { CosmosItemProperties containerDocument = new CosmosItemProperties(); containerDocument.id(markerDocId); - return this.client.createItem(this.leaseCollectionLink, containerDocument, null, false) + return this.client.createItem(this.leaseCollectionLink, containerDocument, new CosmosItemRequestOptions(markerDocId), false) .map( item -> true) .onErrorResume(throwable -> { if (throwable instanceof CosmosClientException) { CosmosClientException e = (CosmosClientException) throwable; if (e.statusCode() == ChangeFeedHelper.HTTP_STATUS_CODE_CONFLICT) { + logger.info("Lease synchronization document was created by a different instance"); return Mono.just(true); } } + logger.error("Unexpected exception thrown", throwable); return Mono.just(false); }); } @@ -91,7 +98,7 @@ public Mono acquireInitializationLock(Duration lockExpirationTime) { containerDocument.id(lockId); BridgeInternal.setProperty(containerDocument, com.azure.data.cosmos.internal.Constants.Properties.TTL, Long.valueOf(lockExpirationTime.getSeconds()).intValue()); - return this.client.createItem(this.leaseCollectionLink, containerDocument, null, false) + return this.client.createItem(this.leaseCollectionLink, containerDocument, new CosmosItemRequestOptions(lockId), false) .map(documentResourceResponse -> { if (documentResourceResponse.item() != null) { this.lockETag = documentResourceResponse.properties().etag(); @@ -104,9 +111,11 @@ public Mono acquireInitializationLock(Duration lockExpirationTime) { if (throwable instanceof CosmosClientException) { CosmosClientException e = (CosmosClientException) throwable; if (e.statusCode() == ChangeFeedHelper.HTTP_STATUS_CODE_CONFLICT) { + logger.info("Lease synchronization document was acquired by a different instance"); return Mono.just(false); } } + logger.error("Unexpected exception thrown", throwable); return Mono.error(throwable); }); } @@ -129,7 +138,7 @@ public Mono releaseInitializationLock() { accessCondition.condition(this.lockETag); requestOptions.accessCondition(accessCondition); - CosmosItem docItem = this.client.getContainerClient().getItem(lockId, "/id"); + CosmosItem docItem = this.client.getContainerClient().getItem(lockId, lockId); return this.client.deleteItem(docItem, requestOptions) .map(documentResourceResponse -> { if (documentResourceResponse.item() != null) { @@ -143,10 +152,11 @@ public Mono releaseInitializationLock() { if (throwable instanceof CosmosClientException) { CosmosClientException e = (CosmosClientException) throwable; if (e.statusCode() == ChangeFeedHelper.HTTP_STATUS_CODE_CONFLICT) { + logger.info("Lease synchronization document was acquired by a different instance"); return Mono.just(false); } } - + logger.error("Unexpected exception thrown", throwable); return Mono.error(throwable); }); } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DocumentServiceLeaseUpdaterImpl.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DocumentServiceLeaseUpdaterImpl.java index e28c13cbf500..257a86b32f2f 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DocumentServiceLeaseUpdaterImpl.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/DocumentServiceLeaseUpdaterImpl.java @@ -12,6 +12,7 @@ import com.azure.data.cosmos.internal.changefeed.Lease; import com.azure.data.cosmos.internal.changefeed.ServiceItemLease; import com.azure.data.cosmos.internal.changefeed.ServiceItemLeaseUpdater; +import com.azure.data.cosmos.internal.changefeed.exceptions.LeaseConflictException; import com.azure.data.cosmos.internal.changefeed.exceptions.LeaseLostException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -78,21 +79,39 @@ public Mono updateLease(Lease cachedLease, CosmosItem itemLink, CosmosIte CosmosItemProperties document = cosmosItemResponse.properties(); ServiceItemLease serverLease = ServiceItemLease.fromDocument(document); logger.info( - "Partition {} update failed because the lease with token '{}' was updated by host '{}' with token '{}'.", + "Partition {} update failed because the lease with token '{}' was updated by owner '{}' with token '{}'.", arrayLease[0].getLeaseToken(), arrayLease[0].getConcurrencyToken(), serverLease.getOwner(), serverLease.getConcurrencyToken()); arrayLease[0] = serverLease; - throw new RuntimeException("Partition update failed"); + throw new LeaseConflictException(arrayLease[0], "Partition update failed"); }); }) .retry(RETRY_COUNT_ON_CONFLICT, throwable -> { - if (throwable instanceof RuntimeException) { - return throwable instanceof LeaseLostException; + if (throwable instanceof LeaseConflictException) { + logger.info( + "Partition {} for the lease with token '{}' failed to update for owner '{}'; will retry.", + arrayLease[0].getLeaseToken(), + arrayLease[0].getConcurrencyToken(), + arrayLease[0].getOwner()); + return true; } return false; + }) + .onErrorResume(throwable -> { + if (throwable instanceof LeaseConflictException) { + logger.warn( + "Partition {} for the lease with token '{}' failed to update for owner '{}'; current continuation token '{}'.", + arrayLease[0].getLeaseToken(), + arrayLease[0].getConcurrencyToken(), + arrayLease[0].getOwner(), + arrayLease[0].getContinuationToken(), throwable); + + return Mono.just(arrayLease[0]); + } + return Mono.error(throwable); }); } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/LeaseRenewerImpl.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/LeaseRenewerImpl.java index cf466b109d03..87328513f8fb 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/LeaseRenewerImpl.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/LeaseRenewerImpl.java @@ -62,9 +62,12 @@ public Mono run(CancellationToken cancellationToken) { return !cancellationToken.isCancellationRequested(); }) .then() - .onErrorResume(throwable -> { - logger.error("Partition {}: renew lease loop failed.", this.lease.getLeaseToken(), throwable); - return Mono.error(throwable); + .doOnError(throwable -> { + if (throwable instanceof LeaseLostException) { + logger.info("Partition {}: renew lease loop failed.", this.lease.getLeaseToken(), throwable); + } else { + logger.error("Partition {}: renew lease loop failed.", this.lease.getLeaseToken(), throwable); + } }); } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/LeaseStoreManagerImpl.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/LeaseStoreManagerImpl.java index 593860803120..4bdf15396de1 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/LeaseStoreManagerImpl.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/LeaseStoreManagerImpl.java @@ -379,7 +379,11 @@ public Mono checkpoint(Lease lease, String continuationToken) { serverLease.setContinuationToken(continuationToken); return serverLease; - })); + })) + .doOnError(throwable -> { + logger.info("Partition {} lease with token '{}' failed to checkpoint for owner '{}' with continuation token '{}'", + lease.getLeaseToken(), lease.getConcurrencyToken(), lease.getOwner(), lease.getContinuationToken()); + }); } @Override @@ -453,6 +457,6 @@ private String getPartitionLeasePrefix() { } private CosmosItem createItemForLease(String leaseId) { - return this.leaseDocumentClient.getContainerClient().getItem(leaseId, "/id"); + return this.leaseDocumentClient.getContainerClient().getItem(leaseId, leaseId); } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/ObserverExceptionWrappingChangeFeedObserverDecorator.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/ObserverExceptionWrappingChangeFeedObserverDecorator.java index 301c5230df7d..7d64811d3871 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/ObserverExceptionWrappingChangeFeedObserverDecorator.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/ObserverExceptionWrappingChangeFeedObserverDecorator.java @@ -32,7 +32,7 @@ public void open(ChangeFeedObserverContext context) { } catch (RuntimeException userException) { - this.logger.warn("Exception happened on ChangeFeedObserver.open", userException); + this.logger.warn("Exception thrown during ChangeFeedObserver.open from thread {}", Thread.currentThread().getId(), userException); throw new ObserverException(userException); } } @@ -44,7 +44,7 @@ public void close(ChangeFeedObserverContext context, ChangeFeedObserverCloseReas } catch (RuntimeException userException) { - this.logger.warn("Exception happened on ChangeFeedObserver.close", userException); + this.logger.warn("Exception thrown during ChangeFeedObserver.close from thread {}", Thread.currentThread().getId(), userException); throw new ObserverException(userException); } } @@ -52,9 +52,8 @@ public void close(ChangeFeedObserverContext context, ChangeFeedObserverCloseReas @Override public Mono processChanges(ChangeFeedObserverContext context, List docs) { return this.changeFeedObserver.processChanges(context, docs) - .onErrorResume(throwable -> { - this.logger.warn("Exception happened on ChangeFeedObserver.processChanges", throwable); - return Mono.error(new ObserverException(throwable)); + .doOnError(throwable -> { + this.logger.warn("Exception thrown during ChangeFeedObserver.processChanges from thread {}", Thread.currentThread().getId(), throwable); }); } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/PartitionCheckpointerImpl.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/PartitionCheckpointerImpl.java index 894f7c17e1d1..b58e20c1755a 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/PartitionCheckpointerImpl.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/PartitionCheckpointerImpl.java @@ -23,13 +23,12 @@ public PartitionCheckpointerImpl(LeaseCheckpointer leaseCheckpointer, Lease leas } @Override - public Mono checkpointPartition(String сontinuationToken) { + public Mono checkpointPartition(String сontinuationToken) { return this.leaseCheckpointer.checkpoint(this.lease, сontinuationToken) .map(lease1 -> { this.lease = lease1; logger.info("Checkpoint: partition {}, new continuation {}", this.lease.getLeaseToken(), this.lease.getContinuationToken()); return lease1; - }) - .then(); + }); } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/PartitionProcessorImpl.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/PartitionProcessorImpl.java index f3a02c8177d4..6684c0bfcdf1 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/PartitionProcessorImpl.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/changefeed/implementation/PartitionProcessorImpl.java @@ -13,6 +13,7 @@ import com.azure.data.cosmos.internal.changefeed.PartitionCheckpointer; import com.azure.data.cosmos.internal.changefeed.PartitionProcessor; import com.azure.data.cosmos.internal.changefeed.ProcessorSettings; +import com.azure.data.cosmos.internal.changefeed.exceptions.LeaseLostException; import com.azure.data.cosmos.internal.changefeed.exceptions.PartitionNotFoundException; import com.azure.data.cosmos.internal.changefeed.exceptions.PartitionSplitException; import com.azure.data.cosmos.internal.changefeed.exceptions.TaskCancelledException; @@ -96,11 +97,14 @@ public Mono run(CancellationToken cancellationToken) { this.lastContinuation = documentFeedResponse.continuationToken(); if (documentFeedResponse.results() != null && documentFeedResponse.results().size() > 0) { return this.dispatchChanges(documentFeedResponse) - .doFinally( (Void) -> { + .doOnError(throwable -> { + logger.debug("Exception was thrown from thread {}", Thread.currentThread().getId(), throwable); + }) + .doOnSuccess((Void) -> { this.options.requestContinuation(this.lastContinuation); if (cancellationToken.isCancellationRequested()) throw new TaskCancelledException(); - }).flux(); + }); } this.options.requestContinuation(this.lastContinuation); @@ -119,7 +123,8 @@ public Mono run(CancellationToken cancellationToken) { if (throwable instanceof CosmosClientException) { CosmosClientException clientException = (CosmosClientException) throwable; - this.logger.warn("Exception: partition {}", this.options.partitionKey().getInternalPartitionKey(), clientException); + logger.warn("CosmosClientException: partition {} from thread {}", + this.options.partitionKey().getInternalPartitionKey(), Thread.currentThread().getId(), clientException); StatusCodeErrorType docDbError = ExceptionClassifier.classifyClientException(clientException); switch (docDbError) { @@ -136,12 +141,12 @@ public Mono run(CancellationToken cancellationToken) { if (this.options.maxItemCount() == null) { this.options.maxItemCount(DefaultMaxItemCount); } else if (this.options.maxItemCount() <= 1) { - this.logger.error("Cannot reduce maxItemCount further as it's already at {}", this.options.maxItemCount(), clientException); + logger.error("Cannot reduce maxItemCount further as it's already at {}", this.options.maxItemCount(), clientException); this.resultException = new RuntimeException(clientException); } this.options.maxItemCount(this.options.maxItemCount() / 2); - this.logger.warn("Reducing maxItemCount, new value: {}", this.options.maxItemCount()); + logger.warn("Reducing maxItemCount, new value: {}", this.options.maxItemCount()); return Flux.empty(); } case TRANSIENT_ERROR: { @@ -157,13 +162,21 @@ public Mono run(CancellationToken cancellationToken) { } } default: { - this.logger.error("Unrecognized DocDbError enum value {}", docDbError, clientException); + logger.error("Unrecognized Cosmos exception returned error code {}", docDbError, clientException); this.resultException = new RuntimeException(clientException); } } + } else if (throwable instanceof LeaseLostException) { + logger.info("LeaseLoseException with partition {} from thread {}", + this.options.partitionKey().getInternalPartitionKey(), Thread.currentThread().getId()); + this.resultException = (LeaseLostException) throwable; } else if (throwable instanceof TaskCancelledException) { - this.logger.debug("Exception: partition {}", this.settings.getPartitionKeyRangeId(), throwable); + logger.debug("Task cancelled exception: partition {} from {}", + this.settings.getPartitionKeyRangeId(), Thread.currentThread().getId(), throwable); this.resultException = (TaskCancelledException) throwable; + } else { + logger.warn("Unexpected exception from thread {}", Thread.currentThread().getId(), throwable); + this.resultException = new RuntimeException(throwable); } return Flux.error(throwable); }) @@ -175,8 +188,13 @@ public Mono run(CancellationToken cancellationToken) { return true; }) - .onErrorResume(throwable -> Flux.empty()) - .then(); + .onErrorResume(throwable -> { + if (this.resultException == null) { + this.resultException = new RuntimeException(throwable); + } + + return Flux.empty(); + }).then(); } @Override diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/ConsistencyWriter.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/ConsistencyWriter.java index 3c51d58fd756..00d196bed6a7 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/ConsistencyWriter.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/ConsistencyWriter.java @@ -170,7 +170,7 @@ Mono writePrivateAsync( try { BridgeInternal.recordResponse(request.requestContext.cosmosResponseDiagnostics, request, storeReader.createStoreResult(null, ex, false, false, primaryUri)); - } catch (CosmosClientException e) { + } catch (Exception e) { logger.error("Error occurred while recording response", e); } String value = ex.responseHeaders().get(HttpConstants.HttpHeaders.WRITE_REQUEST_TRIGGER_ADDRESS_REFRESH); @@ -191,7 +191,7 @@ Mono writePrivateAsync( try { BridgeInternal.recordResponse(request.requestContext.cosmosResponseDiagnostics, request, storeReader.createStoreResult(response, null, false, false, primaryURI.get())); - } catch (CosmosClientException e) { + } catch (Exception e) { logger.error("Error occurred while recording response", e); } return barrierForGlobalStrong(request, response); @@ -309,43 +309,35 @@ private Mono waitForWriteBarrierAsync(RxDocumentServiceRequest barrierR } //get max global committed lsn from current batch of responses, then update if greater than max of all batches. - long maxGlobalCommittedLsn = (responses != null || !responses.isEmpty()) ? - (Long) responses.stream().map(s -> s.globalCommittedLSN).max(ComparatorUtils.NATURAL_COMPARATOR).get() : + long maxGlobalCommittedLsn = (responses != null) ? + (Long) responses.stream().map(s -> s.globalCommittedLSN).max(ComparatorUtils.NATURAL_COMPARATOR).orElse(0L) : 0L; + maxGlobalCommittedLsnReceived.set(maxGlobalCommittedLsnReceived.get() > maxGlobalCommittedLsn ? maxGlobalCommittedLsnReceived.get() : maxGlobalCommittedLsn); //only refresh on first barrier call, set to false for subsequent attempts. barrierRequest.requestContext.forceRefreshAddressCache = false; - //trace on last retry. + //get max global committed lsn from current batch of responses, then update if greater than max of all batches. if (writeBarrierRetryCount.getAndDecrement() == 0) { logger.debug("ConsistencyWriter: WaitForWriteBarrierAsync - Last barrier multi-region strong. Responses: {}", responses.stream().map(StoreResult::toString).collect(Collectors.joining("; "))); + logger.debug("ConsistencyWriter: Highest global committed lsn received for write barrier call is {}", maxGlobalCommittedLsnReceived); + return Mono.just(Boolean.FALSE); } return Mono.empty(); }).flux(); - }).repeatWhen(s -> { - if (writeBarrierRetryCount.get() == 0) { - return Flux.empty(); + }).repeatWhen(s -> s.flatMap(x -> { + // repeat with a delay + if ((ConsistencyWriter.MAX_NUMBER_OF_WRITE_BARRIER_READ_RETRIES - writeBarrierRetryCount.get()) > ConsistencyWriter.MAX_SHORT_BARRIER_RETRIES_FOR_MULTI_REGION) { + return Mono.delay(Duration.ofMillis(ConsistencyWriter.DELAY_BETWEEN_WRITE_BARRIER_CALLS_IN_MS)).flux(); } else { - - if ((ConsistencyWriter.MAX_NUMBER_OF_WRITE_BARRIER_READ_RETRIES - writeBarrierRetryCount.get()) > ConsistencyWriter.MAX_SHORT_BARRIER_RETRIES_FOR_MULTI_REGION) { - return Flux.just(0L).delayElements(Duration.ofMillis(ConsistencyWriter.DELAY_BETWEEN_WRITE_BARRIER_CALLS_IN_MS)); - } else { - return Flux.just(0L).delayElements(Duration.ofMillis(ConsistencyWriter.SHORT_BARRIER_RETRY_INTERVAL_IN_MS_FOR_MULTI_REGION)); - } + return Mono.delay(Duration.ofMillis(ConsistencyWriter.SHORT_BARRIER_RETRY_INTERVAL_IN_MS_FOR_MULTI_REGION)).flux(); } - }).take(1) - .switchIfEmpty(Mono.defer(() -> { - // after retries exhausted print this log and return false - logger.debug("ConsistencyWriter: Highest global committed lsn received for write barrier call is {}", maxGlobalCommittedLsnReceived); - - return Mono.just(false); - })) - .map(r -> r) - .single(); + }) + ).take(1).single(); } static void getLsnAndGlobalCommittedLsn(StoreResponse response, Utils.ValueHolder lsn, Utils.ValueHolder globalCommittedLsn) { diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/StoreReader.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/StoreReader.java index 72e58edfa8d6..1d2dd33be114 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/StoreReader.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/StoreReader.java @@ -248,7 +248,11 @@ private Flux> readFromReplicas(List resultCollect for (StoreResult srr : newStoreResults) { entity.requestContext.requestChargeTracker.addCharge(srr.requestCharge); - BridgeInternal.recordResponse(entity.requestContext.cosmosResponseDiagnostics, entity, srr); + try { + BridgeInternal.recordResponse(entity.requestContext.cosmosResponseDiagnostics, entity, srr); + } catch (Exception e) { + logger.error("Unexpected failure while recording response", e); + } if (srr.isValid) { try { @@ -557,7 +561,11 @@ private Mono readPrimaryInternalAsync( }); return storeResultObs.map(storeResult -> { - BridgeInternal.recordResponse(entity.requestContext.cosmosResponseDiagnostics, entity, storeResult); + try { + BridgeInternal.recordResponse(entity.requestContext.cosmosResponseDiagnostics, entity, storeResult); + } catch (Exception e) { + logger.error("Unexpected failure while recording response", e); + } entity.requestContext.requestChargeTracker.addCharge(storeResult.requestCharge); if (storeResult.isGoneException && !storeResult.isInvalidPartitionException) { diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestManager.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestManager.java index 426242e1804a..04aa6d484018 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestManager.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestManager.java @@ -246,7 +246,7 @@ public void channelUnregistered(final ChannelHandlerContext context) { if (!this.closingExceptionally) { this.completeAllPendingRequestsExceptionally(context, ON_CHANNEL_UNREGISTERED); } else { - logger.warn("{} channelUnregistered exceptionally", context); + logger.debug("{} channelUnregistered exceptionally", context); } context.fireChannelUnregistered(); diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/DocumentQuerySpyWireContentTest.java b/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/DocumentQuerySpyWireContentTest.java index 4ed2b01753e3..7a8c96c1c4bd 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/DocumentQuerySpyWireContentTest.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/DocumentQuerySpyWireContentTest.java @@ -101,7 +101,7 @@ public void queryWithContinuationTokenLimit(FeedOptions options, String query, b .collectList().block(); assertThat(results.size()).describedAs("total results").isGreaterThanOrEqualTo(1); - + List requests = client.getCapturedRequests(); for(HttpRequest req: requests) { @@ -111,17 +111,19 @@ public void queryWithContinuationTokenLimit(FeedOptions options, String query, b private void validateRequestHasContinuationTokenLimit(HttpRequest request, Integer expectedValue) { Map headers = request.headers().toMap(); - if (expectedValue != null && expectedValue > 0) { - assertThat(headers - .containsKey(HttpConstants.HttpHeaders.RESPONSE_CONTINUATION_TOKEN_LIMIT_IN_KB)) + if(headers.get(HttpConstants.HttpHeaders.IS_QUERY) != null ) { + if (expectedValue != null && expectedValue > 0) { + assertThat(headers + .containsKey(HttpConstants.HttpHeaders.RESPONSE_CONTINUATION_TOKEN_LIMIT_IN_KB)) .isTrue(); - assertThat(headers - .get("x-ms-documentdb-responsecontinuationtokenlimitinkb")) + assertThat(headers + .get("x-ms-documentdb-responsecontinuationtokenlimitinkb")) .isEqualTo(Integer.toString(expectedValue)); - } else { - assertThat(headers - .containsKey(HttpConstants.HttpHeaders.RESPONSE_CONTINUATION_TOKEN_LIMIT_IN_KB)) + } else { + assertThat(headers + .containsKey(HttpConstants.HttpHeaders.RESPONSE_CONTINUATION_TOKEN_LIMIT_IN_KB)) .isFalse(); + } } } @@ -159,7 +161,7 @@ public void beforeClass() throws Exception { FeedOptions options = new FeedOptions(); options.enableCrossPartitionQuery(true); - + // do the query once to ensure the collection is cached. client.queryDocuments(getMultiPartitionCollectionLink(), "select * from root", options) .then().block(); diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/SessionTest.java b/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/SessionTest.java index e3253f9f49c8..f74431eb3b71 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/SessionTest.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/SessionTest.java @@ -111,21 +111,17 @@ public void sessionConsistency_ReadYourWrites(boolean isNameBased) { Document documentCreated = spyClient.createDocument(getCollectionLink(isNameBased), new Document(), null, false) .blockFirst().getResource(); - // We send session tokens on Writes in GATEWAY mode - if (connectionMode == ConnectionMode.GATEWAY) { - assertThat(getSessionTokensInRequests()).hasSize(3 * i + 1); - assertThat(getSessionTokensInRequests().get(3 * i + 0)).isNotEmpty(); - } + spyClient.clearCapturedRequests(); spyClient.readDocument(getDocumentLink(documentCreated, isNameBased), options).blockFirst(); - assertThat(getSessionTokensInRequests()).hasSize(3 * i + 2); - assertThat(getSessionTokensInRequests().get(3 * i + 1)).isNotEmpty(); + assertThat(getSessionTokensInRequests()).hasSize(1); + assertThat(getSessionTokensInRequests().get(0)).isNotEmpty(); spyClient.readDocument(getDocumentLink(documentCreated, isNameBased), options).blockFirst(); - assertThat(getSessionTokensInRequests()).hasSize(3 * i + 3); - assertThat(getSessionTokensInRequests().get(3 * i + 2)).isNotEmpty(); + assertThat(getSessionTokensInRequests()).hasSize(2); + assertThat(getSessionTokensInRequests().get(1)).isNotEmpty(); } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/directconnectivity/GatewayAddressCacheTest.java b/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/directconnectivity/GatewayAddressCacheTest.java index 9f43c1c8f914..da3991318195 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/directconnectivity/GatewayAddressCacheTest.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/directconnectivity/GatewayAddressCacheTest.java @@ -385,7 +385,10 @@ public Mono> answer(InvocationOnMock invocationOnMock) throws Thro .describedAs("getServerAddressesViaGatewayAsync will read addresses from gateway") .asList().hasSize(1); httpClientWrapper.capturedRequests.clear(); - assertThat(suboptimalAddresses).hasSize(ServiceConfig.SystemReplicationPolicy.MaxReplicaSetSize - 1); + + // relaxes one replica being down + assertThat(suboptimalAddresses.length).isLessThanOrEqualTo((ServiceConfig.SystemReplicationPolicy.MaxReplicaSetSize - 1)); + assertThat(suboptimalAddresses.length).isGreaterThanOrEqualTo(ServiceConfig.SystemReplicationPolicy.MaxReplicaSetSize - 2); assertThat(fetchCounter.get()).isEqualTo(1); // no refresh, use cache diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/query/DocumentProducerTest.java b/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/query/DocumentProducerTest.java index 062847e91c85..0a2d50d0e1f9 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/query/DocumentProducerTest.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/test/java/com/azure/data/cosmos/internal/query/DocumentProducerTest.java @@ -61,7 +61,7 @@ public class DocumentProducerTest { private final static Logger logger = LoggerFactory.getLogger(DocumentProducerTest.class); - private static final long TIMEOUT = 10000; + private static final long TIMEOUT = 20000; private final static String OrderByPayloadFieldName = "payload"; private final static String OrderByItemsFieldName = "orderByItems"; diff --git a/sdk/cosmos/pom.xml b/sdk/cosmos/pom.xml index 25e7233d34c7..64d84cc40737 100644 --- a/sdk/cosmos/pom.xml +++ b/sdk/cosmos/pom.xml @@ -13,7 +13,7 @@ Licensed under the MIT License. com.microsoft.azure azure-cosmos-parent - 3.3.1 + 3.3.2 pom Microsoft Azure Cosmos DB SQL API From 6feae954894a593fe535b51dadd0b3fd51eb9c3a Mon Sep 17 00:00:00 2001 From: milismsft-mac Date: Thu, 7 Nov 2019 10:41:52 -0800 Subject: [PATCH 2/2] update version --- .../main/java/com/azure/data/cosmos/internal/HttpConstants.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/HttpConstants.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/HttpConstants.java index 76d0f627c6f8..7ae29389bfba 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/HttpConstants.java +++ b/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"; }