Skip to content

Cosmos: CFP fixes and port consistency policy bug fix #6180

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
milismsft merged 1 commit into from
Nov 7, 2019
Merged

Cosmos: CFP fixes and port consistency policy bug fix #6180

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion ...mos/azure-cosmos-examples/src/test/java/com/azure/cosmos/rx/examples/ConflictAPITest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,10 +81,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)
Expand Down
4 changes: 3 additions & 1 deletion ...cosmos-examples/src/test/java/com/azure/cosmos/rx/examples/DocumentQueryAsyncAPITest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
5 changes: 4 additions & 1 deletion ...azure-cosmos-examples/src/test/java/com/azure/cosmos/rx/examples/InMemoryGroupbyTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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++) {

Expand All @@ -81,11 +83,12 @@ 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();
tasks.add(client.createDocument(getCollectionLink(), doc, null, true).then());

Thread.sleep(100);
}
}
Flux.merge(tasks).then().block();
System.out.println("finished inserting documents");
}

Expand Down
3 changes: 2 additions & 1 deletion ...s/src/main/java/com/azure/cosmos/internal/changefeed/implementation/BootstrapperImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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 -> {
Expand Down
19 changes: 15 additions & 4 deletions ...n/java/com/azure/cosmos/internal/changefeed/implementation/DocumentServiceLeaseStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
import com.azure.cosmos.internal.changefeed.LeaseStore;
import com.azure.cosmos.internal.changefeed.RequestOptionsFactory;
import com.azure.cosmos.internal.changefeed.ServiceItemLease;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;

import java.time.Duration;
Expand All @@ -23,6 +25,7 @@
* Implementation for LeaseStore.
*/
class DocumentServiceLeaseStore implements LeaseStore {
private final Logger logger = LoggerFactory.getLogger(BootstrapperImpl.class);
private ChangeFeedContextClient client;
private String containerNamePrefix;
private CosmosAsyncContainer leaseCollectionLink;
Expand Down Expand Up @@ -52,16 +55,18 @@ public Mono<Boolean> isInitialized() {
CosmosItemRequestOptions requestOptions = this.requestOptionsFactory.createRequestOptions(
ServiceItemLease.fromDocument(doc));

CosmosAsyncItem docItem = this.client.getContainerClient().getItem(markerDocId, "/id");
CosmosAsyncItem docItem = this.client.getContainerClient().getItem(markerDocId, markerDocId);
return this.client.readItem(docItem, requestOptions)
.flatMap(documentResourceResponse -> Mono.just(documentResourceResponse.getItem() != null))
.onErrorResume(throwable -> {
if (throwable instanceof CosmosClientException) {
CosmosClientException e = (CosmosClientException) throwable;
if (e.getStatusCode() == 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);
});
}
Expand All @@ -72,15 +77,17 @@ public Mono<Boolean> markInitialized() {
CosmosItemProperties containerDocument = new CosmosItemProperties();
containerDocument.setId(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.getStatusCode() == 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);
});
}
Expand All @@ -92,7 +99,7 @@ public Mono<Boolean> acquireInitializationLock(Duration lockExpirationTime) {
containerDocument.setId(lockId);
BridgeInternal.setProperty(containerDocument, 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.getItem() != null) {
this.lockETag = documentResourceResponse.getProperties().getETag();
Expand All @@ -105,9 +112,11 @@ public Mono<Boolean> acquireInitializationLock(Duration lockExpirationTime) {
if (throwable instanceof CosmosClientException) {
CosmosClientException e = (CosmosClientException) throwable;
if (e.getStatusCode() == 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);
});
}
Expand All @@ -130,7 +139,7 @@ public Mono<Boolean> releaseInitializationLock() {
accessCondition.setCondition(this.lockETag);
requestOptions.setAccessCondition(accessCondition);

CosmosAsyncItem docItem = this.client.getContainerClient().getItem(lockId, "/id");
CosmosAsyncItem docItem = this.client.getContainerClient().getItem(lockId, lockId);
return this.client.deleteItem(docItem, requestOptions)
.map(documentResourceResponse -> {
if (documentResourceResponse.getItem() != null) {
Expand All @@ -144,10 +153,12 @@ public Mono<Boolean> releaseInitializationLock() {
if (throwable instanceof CosmosClientException) {
CosmosClientException e = (CosmosClientException) throwable;
if (e.getStatusCode() == 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);
});
}
Expand Down
6 changes: 5 additions & 1 deletion ...s/src/main/java/com/azure/cosmos/internal/changefeed/implementation/LeaseRenewerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,11 @@ public Mono<Void> run(CancellationToken cancellationToken) {
})
.then()
.onErrorResume(throwable -> {
logger.error("Partition {}: renew lease loop failed.", this.lease.getLeaseToken(), 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);
}
return Mono.error(throwable);
});
}
Expand Down
2 changes: 1 addition & 1 deletion .../main/java/com/azure/cosmos/internal/changefeed/implementation/LeaseStoreManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -453,6 +453,6 @@ private String getPartitionLeasePrefix() {
}

private CosmosAsyncItem createItemForLease(String leaseId) {
return this.leaseDocumentClient.getContainerClient().getItem(leaseId, "/id");
return this.leaseDocumentClient.getContainerClient().getItem(leaseId, leaseId);
}
}
37 changes: 14 additions & 23 deletions ...-cosmos/src/main/java/com/azure/cosmos/internal/directconnectivity/ConsistencyWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ Mono<StoreResponse> 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.getResponseHeaders().get(HttpConstants.HttpHeaders.WRITE_REQUEST_TRIGGER_ADDRESS_REFRESH);
Expand All @@ -191,7 +191,7 @@ Mono<StoreResponse> 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);
Expand Down Expand Up @@ -309,43 +309,34 @@ private Mono<Boolean> 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<Long> lsn, Utils.ValueHolder<Long> globalCommittedLsn) {
Expand Down
12 changes: 10 additions & 2 deletions .../azure-cosmos/src/main/java/com/azure/cosmos/internal/directconnectivity/StoreReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,7 +248,11 @@ private Flux<List<StoreResult>> readFromReplicas(List<StoreResult> 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 {
Expand Down Expand Up @@ -557,7 +561,11 @@ private Mono<ReadReplicaResult> 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) {
Expand Down
14 changes: 5 additions & 9 deletions sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/internal/SessionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
}

Expand Down
5 changes: 4 additions & 1 deletion ...s/src/test/java/com/azure/cosmos/internal/directconnectivity/GatewayAddressCacheTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -396,7 +396,10 @@ public Mono<List<Address>> answer(InvocationOnMock invocationOnMock) throws Thro
.describedAs("getServerAddressesViaGatewayAsync will read addresses from gateway")
.asList().hasSize(1);
httpClientWrapper.capturedRequests.clear();
AssertionsForClassTypes.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
Expand Down
2 changes: 1 addition & 1 deletion ...smos/azure-cosmos/src/test/java/com/azure/cosmos/internal/query/DocumentProducerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,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";

Expand Down
2 changes: 1 addition & 1 deletion ...cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/DocumentClientResourceLeakTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ public void resourceLeak() throws Exception {
usedMemoryInBytesBefore / (double)ONE_MB,
(usedMemoryInBytesAfter - usedMemoryInBytesBefore) / (double)ONE_MB);

assertThat(usedMemoryInBytesAfter - usedMemoryInBytesBefore).isLessThan(125 * ONE_MB);
assertThat(usedMemoryInBytesAfter - usedMemoryInBytesBefore).isLessThan(300 * ONE_MB);
}

@BeforeClass(groups = {"emulator"}, timeOut = SETUP_TIMEOUT)
Expand Down
4 changes: 4 additions & 0 deletions sdk/cosmos/changelog/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,10 @@
- Namespace changes from microsoft-azure-cosmos to azure-cosmos
- Getter and Setter changes to get() & set() pattern

### 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)

### 3.3.1
- Added @JsonIgnore on getLogger in JsonSerializable

Expand Down