diff --git a/sdk/cosmos/azure-cosmos-tests/pom.xml b/sdk/cosmos/azure-cosmos-tests/pom.xml index 18cb8e9cd2d1..4ee203075178 100644 --- a/sdk/cosmos/azure-cosmos-tests/pom.xml +++ b/sdk/cosmos/azure-cosmos-tests/pom.xml @@ -776,6 +776,33 @@ Licensed under the MIT License. + + + consistency-overrides + + consistency-overrides + + + + + org.apache.maven.plugins + maven-failsafe-plugin + 3.5.3 + + + src/test/resources/consistency-overrides-testng.xml + + + true + 1 + 256 + paranoid + + + + + + emulator @@ -848,6 +875,7 @@ Licensed under the MIT License. true + true 1 256 paranoid diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosConsistencyOverrideValidationTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosConsistencyOverrideValidationTest.java new file mode 100644 index 000000000000..63b5f2d1cce1 --- /dev/null +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosConsistencyOverrideValidationTest.java @@ -0,0 +1,636 @@ +/* + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. + * + */ + +package com.azure.cosmos; + +import com.azure.cosmos.implementation.DatabaseAccount; +import com.azure.cosmos.implementation.DatabaseAccountLocation; +import com.azure.cosmos.implementation.GlobalEndpointManager; +import com.azure.cosmos.implementation.HttpConstants; +import com.azure.cosmos.implementation.RxDocumentClientImpl; +import com.azure.cosmos.implementation.TestConfigurations; +import com.azure.cosmos.implementation.directconnectivity.Protocol; +import com.azure.cosmos.implementation.directconnectivity.ReflectionUtils; +import com.azure.cosmos.models.CosmosChangeFeedRequestOptions; +import com.azure.cosmos.models.CosmosItemRequestOptions; +import com.azure.cosmos.models.CosmosItemResponse; +import com.azure.cosmos.models.CosmosQueryRequestOptions; +import com.azure.cosmos.models.FeedRange; +import com.azure.cosmos.models.FeedResponse; +import com.azure.cosmos.models.PartitionKey; +import com.azure.cosmos.models.SqlParameter; +import com.azure.cosmos.models.SqlQuerySpec; +import com.azure.cosmos.rx.TestSuiteBase; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.testng.SkipException; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Factory; +import org.testng.annotations.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +public class CosmosConsistencyOverrideValidationTest extends TestSuiteBase { + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final String MULTIPLE_WRITE_REGIONS_PROPERTY = + "COSMOS.CONSISTENCY_OVERRIDE_MULTIPLE_WRITE_REGIONS_ENABLED"; + private static final String EMULATOR_VNEXT_ENABLED_PROPERTY = "COSMOS.EMULATOR_VNEXT_ENABLED"; + private static final String HTTP2_ENABLED_PROPERTY = "COSMOS.HTTP2_ENABLED"; + private static final String THINCLIENT_ENABLED_PROPERTY = "COSMOS.THINCLIENT_ENABLED"; + + private CosmosClient client; + private CosmosContainer container; + private final String modeLabel; + private final boolean http2Enabled; + private final boolean thinClientEnabled; + private String previousHttp2Enabled; + private String previousThinClientEnabled; + private ConsistencyLevel accountDefaultConsistency; + + @Factory(dataProvider = "clientBuildersForConsistencyOverrides") + public CosmosConsistencyOverrideValidationTest( + CosmosClientBuilder clientBuilder, + String modeLabel, + boolean http2Enabled, + boolean thinClientEnabled) { + + super(clientBuilder); + this.modeLabel = modeLabel; + this.http2Enabled = http2Enabled; + this.thinClientEnabled = thinClientEnabled; + } + + @Override + public String resolveTestNameSuffix(Object[] row) { + return this.modeLabel; + } + + @DataProvider + public static Object[][] clientBuildersForConsistencyOverrides() { + boolean multipleWriteRegionsEnabled = Boolean.parseBoolean( + System.getProperty(MULTIPLE_WRITE_REGIONS_PROPERTY, "false")); + boolean emulatorVNextRun = isEmulatorVNextRun(); + + List providers = new ArrayList<>(); + if (!emulatorVNextRun) { + addDirectClientBuilder(providers, multipleWriteRegionsEnabled); + } + + addGatewayClientBuilder( + providers, + TestConfigurations.HOST, + multipleWriteRegionsEnabled, + false, + false); + + if (TestConfigurations.HOST.contains(ROUTING_GATEWAY_EMULATOR_PORT) && !emulatorVNextRun) { + String computeGatewayEndpoint = TestConfigurations.HOST.replace( + ROUTING_GATEWAY_EMULATOR_PORT, + COMPUTE_GATEWAY_EMULATOR_PORT); + addGatewayClientBuilder( + providers, + computeGatewayEndpoint, + multipleWriteRegionsEnabled, + false, + false); + } + + if (!emulatorVNextRun && !isEmulatorGatewayEndpoint(TestConfigurations.HOST)) { + addGatewayClientBuilder( + providers, + TestConfigurations.HOST, + multipleWriteRegionsEnabled, + true, + true); + } + + return providers.toArray(new Object[0][]); + } + + @BeforeClass(groups = { "consistency-overrides", "emulator", "emulator-vnext" }, timeOut = SETUP_TIMEOUT) + public void beforeClass() { + assertThat(this.client).isNull(); + + applyTransportSystemProperties(); + this.client = getClientBuilder().buildClient(); + CosmosAsyncContainer asyncContainer = getSharedMultiPartitionCosmosContainer(this.client.asyncClient()); + this.container = this.client + .getDatabase(asyncContainer.getDatabase().getId()) + .getContainer(asyncContainer.getId()); + + DatabaseAccount databaseAccount = getLatestDatabaseAccount(); + this.accountDefaultConsistency = databaseAccount.getConsistencyPolicy().getDefaultConsistencyLevel(); + + logger.info( + "Consistency override test mode [{}], endpoint [{}], HTTP/2 enabled [{}], thin client enabled [{}], account default consistency [{}]", + this.modeLabel, + getClientBuilder().getEndpoint(), + this.http2Enabled, + this.thinClientEnabled, + this.accountDefaultConsistency); + + logAccountMetadata(databaseAccount); + } + + @AfterClass(groups = { "consistency-overrides", "emulator", "emulator-vnext" }, timeOut = SHUTDOWN_TIMEOUT, alwaysRun = true) + public void afterClass() { + safeCloseSyncClient(this.client); + this.client = null; + this.container = null; + this.accountDefaultConsistency = null; + restoreTransportSystemProperties(); + } + + @Test(groups = { "consistency-overrides", "emulator", "emulator-vnext" }, timeOut = TIMEOUT) + public void requestOptionsConsistencyUpgradeReadAndQueryShouldBeIgnored() { + List unsupportedRequestConsistencies = strongerConsistencyLevelsThan(this.accountDefaultConsistency); + if (unsupportedRequestConsistencies.isEmpty()) { + throw new SkipException( + "No request-level consistency upgrade exists for account default " + this.accountDefaultConsistency); + } + + TestItem item = createTestItem(); + List unexpectedOutcomes = new ArrayList<>(); + for (ConsistencyLevel requestedConsistency : unsupportedRequestConsistencies) { + verifyUpgradeIgnoredOrRecord( + "readItem", + requestedConsistency, + () -> executeRead(item, new CosmosItemRequestOptions().setConsistencyLevel(requestedConsistency)), + unexpectedOutcomes); + + verifyUpgradeIgnoredOrRecord( + "pointQuery", + requestedConsistency, + () -> executePointQuery(item, new CosmosQueryRequestOptions().setConsistencyLevel(requestedConsistency)), + unexpectedOutcomes); + } + + assertThat(unexpectedOutcomes) + .as( + "All request-level consistency upgrades should be ignored for mode [%s], endpoint [%s], " + + "account default consistency [%s]", + this.modeLabel, + getClientBuilder().getEndpoint(), + this.accountDefaultConsistency) + .isEmpty(); + } + + @Test(groups = { "consistency-overrides", "emulator", "emulator-vnext" }, timeOut = TIMEOUT) + public void latestCommittedReadConsistencyStrategyReadAndQueryShouldSucceed() { + TestItem item = createTestItem(); + + OperationResult readResult = executeRead( + item, + new CosmosItemRequestOptions().setReadConsistencyStrategy(ReadConsistencyStrategy.LATEST_COMMITTED)); + assertThat(readResult.statusCode).isEqualTo(HttpConstants.StatusCodes.OK); + assertLatestCommitted(readResult, "readItem"); + + OperationResult queryResult = executePointQuery( + item, + new CosmosQueryRequestOptions().setReadConsistencyStrategy(ReadConsistencyStrategy.LATEST_COMMITTED)); + assertThat(queryResult.resultCount).isEqualTo(1); + assertLatestCommitted(queryResult, "pointQuery"); + + OperationResult changeFeedResult = executeChangeFeed( + CosmosChangeFeedRequestOptions + .createForProcessingFromNow(FeedRange.forFullRange()) + .setReadConsistencyStrategy(ReadConsistencyStrategy.LATEST_COMMITTED)); + assertThat(changeFeedResult.statusCode) + .as("queryChangeFeed status code") + .isIn(HttpConstants.StatusCodes.OK, HttpConstants.StatusCodes.NOT_MODIFIED); + assertThat(changeFeedResult.resultCount).isEqualTo(0); + assertLatestCommitted(changeFeedResult, "queryChangeFeed"); + } + + private TestItem createTestItem() { + String id = UUID.randomUUID().toString(); + String partitionKey = UUID.randomUUID().toString(); + ObjectNode item = OBJECT_MAPPER.createObjectNode(); + item.put("id", id); + item.put("mypk", partitionKey); + item.put("payload", UUID.randomUUID().toString()); + + CosmosItemResponse createResponse = this.container.createItem(item); + assertThat(createResponse.getStatusCode()).isEqualTo(HttpConstants.StatusCodes.CREATED); + waitIfNeededForReplicasToCatchUp(); + return new TestItem(id, partitionKey); + } + + private void waitIfNeededForReplicasToCatchUp() { + switch (this.accountDefaultConsistency) { + case EVENTUAL: + case CONSISTENT_PREFIX: + logger.info(" additional wait in EVENTUAL mode so the replica catch up"); + try { + Thread.sleep(WAIT_REPLICA_CATCH_UP_IN_MILLIS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + break; + + default: + break; + } + } + + private OperationResult executeRead(TestItem item, CosmosItemRequestOptions requestOptions) { + CosmosItemResponse response = this.container.readItem( + item.id, + new PartitionKey(item.partitionKey), + requestOptions, + ObjectNode.class); + + return new OperationResult( + response.getDiagnostics().getDiagnosticsContext(), + 1, + 0, + response.getStatusCode()); + } + + private OperationResult executePointQuery(TestItem item, CosmosQueryRequestOptions requestOptions) { + SqlQuerySpec querySpec = new SqlQuerySpec( + "SELECT * FROM c WHERE c.mypk = @pk AND c.id = @id", + new SqlParameter("@pk", item.partitionKey), + new SqlParameter("@id", item.id)); + + Iterator> iterator = this.container + .queryItems(querySpec, requestOptions, ObjectNode.class) + .iterableByPage() + .iterator(); + + int resultCount = 0; + int pageCount = 0; + CosmosDiagnosticsContext diagnosticsContext = null; + while (iterator.hasNext()) { + FeedResponse response = iterator.next(); + pageCount++; + resultCount += response.getResults().size(); + diagnosticsContext = response.getCosmosDiagnostics().getDiagnosticsContext(); + + if (resultCount > 0) { + break; + } + } + + return new OperationResult( + diagnosticsContext, + resultCount, + pageCount, + diagnosticsContext != null ? diagnosticsContext.getStatusCode() : -1); + } + + private OperationResult executeChangeFeed(CosmosChangeFeedRequestOptions requestOptions) { + Iterator> iterator = this.container + .queryChangeFeed(requestOptions, ObjectNode.class) + .iterableByPage() + .iterator(); + + if (!iterator.hasNext()) { + return new OperationResult(null, 0, 0, -1); + } + + FeedResponse response = iterator.next(); + CosmosDiagnosticsContext diagnosticsContext = response.getCosmosDiagnostics().getDiagnosticsContext(); + return new OperationResult( + diagnosticsContext, + response.getResults().size(), + 1, + diagnosticsContext != null ? diagnosticsContext.getStatusCode() : -1); + } + + private void verifyUpgradeIgnoredOrRecord( + String operationName, + ConsistencyLevel requestedConsistency, + ConsistencyOverrideOperation operation, + List unexpectedOutcomes) { + + try { + OperationResult result = operation.execute(); + if (result.statusCode != HttpConstants.StatusCodes.OK || result.resultCount != 1) { + logger.warn( + "CONSISTENCY_OVERRIDE_RESULT mode [{}], endpoint [{}], operation [{}], requested consistency [{}], " + + "outcome [UNEXPECTED_SUCCESS_SHAPE], status [{}], result count [{}], page count [{}], " + + "diagnostics [{}]", + this.modeLabel, + getClientBuilder().getEndpoint(), + operationName, + requestedConsistency, + result.statusCode, + result.resultCount, + result.pageCount, + diagnosticsSummary(result.diagnosticsContext)); + unexpectedOutcomes.add(String.format( + "%s returned unexpected success shape for requested consistency [%s]. Status [%s], " + + "result count [%s], page count [%s], diagnostics [%s]", + operationName, + requestedConsistency, + result.statusCode, + result.resultCount, + result.pageCount, + diagnosticsSummary(result.diagnosticsContext))); + return; + } + + if (result.diagnosticsContext == null + || result.diagnosticsContext.getEffectiveConsistencyLevel() != this.accountDefaultConsistency) { + logger.warn( + "CONSISTENCY_OVERRIDE_RESULT mode [{}], endpoint [{}], operation [{}], requested consistency [{}], " + + "outcome [UNEXPECTED_EFFECTIVE_CONSISTENCY], status [{}], result count [{}], " + + "page count [{}], account default consistency [{}], diagnostics [{}]", + this.modeLabel, + getClientBuilder().getEndpoint(), + operationName, + requestedConsistency, + result.statusCode, + result.resultCount, + result.pageCount, + this.accountDefaultConsistency, + diagnosticsSummary(result.diagnosticsContext)); + unexpectedOutcomes.add(String.format( + "%s did not ignore requested consistency [%s]. Account default consistency [%s], diagnostics [%s]", + operationName, + requestedConsistency, + this.accountDefaultConsistency, + diagnosticsSummary(result.diagnosticsContext))); + return; + } + + logger.info( + "CONSISTENCY_OVERRIDE_RESULT mode [{}], endpoint [{}], operation [{}], requested consistency [{}], " + + "outcome [EXPECTED_IGNORED_UPGRADE], status [{}], result count [{}], page count [{}], " + + "diagnostics [{}]", + this.modeLabel, + getClientBuilder().getEndpoint(), + operationName, + requestedConsistency, + result.statusCode, + result.resultCount, + result.pageCount, + diagnosticsSummary(result.diagnosticsContext)); + } catch (CosmosException error) { + logger.warn( + "CONSISTENCY_OVERRIDE_RESULT mode [{}], endpoint [{}], operation [{}], requested consistency [{}], " + + "outcome [UNEXPECTED_FAILURE], status [{}], substatus [{}], diagnostics [{}]", + this.modeLabel, + getClientBuilder().getEndpoint(), + operationName, + requestedConsistency, + error.getStatusCode(), + error.getSubStatusCode(), + diagnosticsSummary(error.getDiagnostics() != null + ? error.getDiagnostics().getDiagnosticsContext() + : null)); + unexpectedOutcomes.add(String.format( + "%s failed for requested consistency [%s]. Status [%s], substatus [%s], diagnostics [%s], " + + "message [%s]", + operationName, + requestedConsistency, + error.getStatusCode(), + error.getSubStatusCode(), + diagnosticsSummary(error.getDiagnostics() != null + ? error.getDiagnostics().getDiagnosticsContext() + : null), + error.getMessage())); + } + } + + private void assertLatestCommitted(OperationResult result, String operationName) { + assertThat(result.diagnosticsContext) + .as("%s diagnostics should not be null", operationName) + .isNotNull(); + assertThat(result.diagnosticsContext.getEffectiveReadConsistencyStrategy()) + .as("%s effective read consistency strategy", operationName) + .isEqualTo(ReadConsistencyStrategy.LATEST_COMMITTED); + } + + private DatabaseAccount getLatestDatabaseAccount() { + GlobalEndpointManager globalEndpointManager = ReflectionUtils.getGlobalEndpointManager( + (RxDocumentClientImpl) this.client.asyncClient().getContextClient()); + return globalEndpointManager.getLatestDatabaseAccount(); + } + + private void logAccountMetadata(DatabaseAccount databaseAccount) { + logger.info( + "Consistency override test account metadata: multiple write locations enabled [{}], writable regions [{}], readable regions [{}]", + databaseAccount.getEnableMultipleWriteLocations(), + locationNames(databaseAccount.getWritableLocations()), + locationNames(databaseAccount.getReadableLocations())); + } + + private static List locationNames(Iterable locations) { + List names = new ArrayList<>(); + for (DatabaseAccountLocation location : locations) { + names.add(location.getName()); + } + + return names; + } + + private static List strongerConsistencyLevelsThan(ConsistencyLevel accountDefaultConsistency) { + return Arrays.stream(ConsistencyLevel.values()) + .filter(consistencyLevel -> strength(consistencyLevel) > strength(accountDefaultConsistency)) + .collect(Collectors.toList()); + } + + private static int strength(ConsistencyLevel consistencyLevel) { + switch (consistencyLevel) { + case STRONG: + return 5; + case BOUNDED_STALENESS: + return 4; + case SESSION: + return 3; + case CONSISTENT_PREFIX: + return 2; + case EVENTUAL: + return 1; + default: + throw new IllegalArgumentException("Unknown consistency level " + consistencyLevel); + } + } + + private static String diagnosticsSummary(CosmosDiagnosticsContext diagnosticsContext) { + if (diagnosticsContext == null) { + return "null"; + } + + return String.format( + "status=%s, subStatus=%s, effectiveConsistency=%s, effectiveReadConsistencyStrategy=%s", + diagnosticsContext.getStatusCode(), + diagnosticsContext.getSubStatusCode(), + diagnosticsContext.getEffectiveConsistencyLevel(), + diagnosticsContext.getEffectiveReadConsistencyStrategy()); + } + + private static void addGatewayClientBuilder( + List providers, + String endpoint, + boolean multipleWriteRegionsEnabled, + boolean http2Enabled, + boolean thinClientEnabled) { + + providers.add(new Object[] { + createGatewayBuilder(endpoint, multipleWriteRegionsEnabled, http2Enabled, thinClientEnabled), + modeLabel(endpoint, multipleWriteRegionsEnabled, http2Enabled, thinClientEnabled), + http2Enabled, + thinClientEnabled + }); + } + + private static void addDirectClientBuilder(List providers, boolean multipleWriteRegionsEnabled) { + CosmosClientBuilder builder = createDirectRxDocumentClient( + null, + Protocol.TCP, + multipleWriteRegionsEnabled, + preferredLocations, + true, + true); + builder.multipleWriteRegionsEnabled(multipleWriteRegionsEnabled); + + providers.add(new Object[] { + builder, + "DirectTcp" + (multipleWriteRegionsEnabled ? "-MultiWrite" : "-SingleWrite"), + false, + false + }); + } + + private static CosmosClientBuilder createGatewayBuilder( + String endpoint, + boolean multipleWriteRegionsEnabled, + boolean http2Enabled, + boolean thinClientEnabled) { + + String previousHttp2Enabled = System.getProperty(HTTP2_ENABLED_PROPERTY); + String previousThinClientEnabled = System.getProperty(THINCLIENT_ENABLED_PROPERTY); + + try { + setTransportSystemProperties(http2Enabled, thinClientEnabled); + return createGatewayRxDocumentClient( + endpoint, + null, + multipleWriteRegionsEnabled, + preferredLocations, + true, + true, + http2Enabled); + } finally { + restoreSystemProperty(HTTP2_ENABLED_PROPERTY, previousHttp2Enabled); + restoreSystemProperty(THINCLIENT_ENABLED_PROPERTY, previousThinClientEnabled); + } + } + + private void applyTransportSystemProperties() { + this.previousHttp2Enabled = System.getProperty(HTTP2_ENABLED_PROPERTY); + this.previousThinClientEnabled = System.getProperty(THINCLIENT_ENABLED_PROPERTY); + setTransportSystemProperties(this.http2Enabled, this.thinClientEnabled); + } + + private void restoreTransportSystemProperties() { + restoreSystemProperty(HTTP2_ENABLED_PROPERTY, this.previousHttp2Enabled); + restoreSystemProperty(THINCLIENT_ENABLED_PROPERTY, this.previousThinClientEnabled); + this.previousHttp2Enabled = null; + this.previousThinClientEnabled = null; + } + + private static void setTransportSystemProperties(boolean http2Enabled, boolean thinClientEnabled) { + System.setProperty(HTTP2_ENABLED_PROPERTY, Boolean.toString(http2Enabled)); + System.setProperty(THINCLIENT_ENABLED_PROPERTY, Boolean.toString(thinClientEnabled)); + } + + private static void restoreSystemProperty(String propertyName, String previousValue) { + if (previousValue == null) { + System.clearProperty(propertyName); + } else { + System.setProperty(propertyName, previousValue); + } + } + + private static boolean isEmulatorGatewayEndpoint(String endpoint) { + return endpoint.contains(ROUTING_GATEWAY_EMULATOR_PORT) + || endpoint.contains(COMPUTE_GATEWAY_EMULATOR_PORT); + } + + private static boolean isEmulatorVNextRun() { + return Boolean.parseBoolean(System.getProperty(EMULATOR_VNEXT_ENABLED_PROPERTY, "false")) + || containsEmulatorVNextGroup(System.getProperty("test.groups")) + || containsEmulatorVNextGroup(System.getProperty("groups")) + || containsEmulatorVNextGroup(System.getProperty("includedGroups")) + || TestConfigurations.HOST.startsWith("http://"); + } + + private static boolean containsEmulatorVNextGroup(String groups) { + return groups != null && Arrays.stream(groups.split(",")) + .map(String::trim) + .anyMatch("emulator-vnext"::equals); + } + + private static String modeLabel( + String endpoint, + boolean multipleWriteRegionsEnabled, + boolean http2Enabled, + boolean thinClientEnabled) { + + String gatewayLabel; + if (endpoint.contains(COMPUTE_GATEWAY_EMULATOR_PORT)) { + gatewayLabel = "ComputeGateway"; + } else if (endpoint.contains(ROUTING_GATEWAY_EMULATOR_PORT)) { + gatewayLabel = "RoutingGateway"; + } else if (thinClientEnabled) { + gatewayLabel = "GatewayV2ThinClient"; + } else if (http2Enabled) { + gatewayLabel = "GatewayV2Http2"; + } else { + gatewayLabel = "GatewayV1"; + } + + return gatewayLabel + (multipleWriteRegionsEnabled ? "-MultiWrite" : "-SingleWrite"); + } + + @FunctionalInterface + private interface ConsistencyOverrideOperation { + OperationResult execute(); + } + + private static final class OperationResult { + private final CosmosDiagnosticsContext diagnosticsContext; + private final int resultCount; + private final int pageCount; + private final int statusCode; + + private OperationResult( + CosmosDiagnosticsContext diagnosticsContext, + int resultCount, + int pageCount, + int statusCode) { + + this.diagnosticsContext = diagnosticsContext; + this.resultCount = resultCount; + this.pageCount = pageCount; + this.statusCode = statusCode; + } + } + + private static final class TestItem { + private final String id; + private final String partitionKey; + + private TestItem(String id, String partitionKey) { + this.id = id; + this.partitionKey = partitionKey; + } + } +} diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/RxDocumentClientImplTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/RxDocumentClientImplTest.java index cbc9301142f4..1519713a68f9 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/RxDocumentClientImplTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/RxDocumentClientImplTest.java @@ -16,6 +16,7 @@ import com.azure.cosmos.implementation.caches.RxClientCollectionCache; import com.azure.cosmos.implementation.caches.RxPartitionKeyRangeCache; import com.azure.cosmos.implementation.directconnectivity.ReflectionUtils; +import com.azure.cosmos.implementation.directconnectivity.GatewayServiceConfigurationReader; import com.azure.cosmos.implementation.directconnectivity.StoreResponse; import com.azure.cosmos.implementation.http.HttpClient; import com.azure.cosmos.implementation.http.HttpClientConfig; @@ -123,6 +124,7 @@ public void readMany() { // setup mocks DocumentClientRetryPolicy documentClientRetryPolicyMock = Mockito.mock(DocumentClientRetryPolicy.class); + GatewayServiceConfigurationReader gatewayServiceConfigurationReaderMock = Mockito.mock(GatewayServiceConfigurationReader.class); RxGatewayStoreModel gatewayStoreModelMock = Mockito.mock(RxGatewayStoreModel.class); RxStoreModel serverStoreModelMock = Mockito.mock(RxStoreModel.class); @@ -223,6 +225,10 @@ public void readMany() { Mockito.when(this.cosmosAuthorizationTokenResolverMock.getAuthorizationToken(Mockito.anyString(), Mockito.anyString(), Mockito.anyString(), Mockito.any())).thenReturn("abcdefgh"); Mockito.when(this.resetSessionTokenRetryPolicyMock.getRequestPolicy(Mockito.any())).thenReturn(documentClientRetryPolicyMock); Mockito.when(documentClientRetryPolicyMock.getRetryContext()).thenReturn(retryContext); + Mockito.when(documentClientRetryPolicyMock.shouldRetry(Mockito.any(Exception.class))) + .thenReturn(Mono.just(ShouldRetryResult.noRetry())); + Mockito.when(gatewayServiceConfigurationReaderMock.getDefaultConsistencyLevel()) + .thenReturn(ConsistencyLevel.SESSION); Mockito .when(serverStoreModelMock.processMessage(Mockito.any(RxDocumentServiceRequest.class))) .thenReturn(Mono.just(mockRxDocumentServiceResponse(pointReadResult, headersForPointReads))); @@ -256,6 +262,7 @@ public void readMany() { ReflectionUtils.setCollectionCache(rxDocumentClient, this.collectionCacheMock); ReflectionUtils.setPartitionKeyRangeCache(rxDocumentClient, this.partitionKeyRangeCacheMock); ReflectionUtils.setResetSessionTokenRetryPolicy(rxDocumentClient, this.resetSessionTokenRetryPolicyMock); + ReflectionUtils.setGatewayServiceConfigurationReader(rxDocumentClient, gatewayServiceConfigurationReaderMock); ReflectionUtils.setGatewayProxy(rxDocumentClient, gatewayStoreModelMock); ReflectionUtils.setServerStoreModel(rxDocumentClient, serverStoreModelMock); @@ -431,7 +438,7 @@ public void onBeforeSendRequest(RxDocumentServiceRequest request) {} @Override public Mono shouldRetry(Exception e) { - return null; + return Mono.just(ShouldRetryResult.noRetry()); } @Override diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/ReflectionUtils.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/ReflectionUtils.java index d657eab12e5b..39d720e2abc3 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/ReflectionUtils.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/ReflectionUtils.java @@ -269,6 +269,12 @@ public static void setGatewayProxy(RxDocumentClientImpl client, RxStoreModel sto set(client, storeModel, "gatewayProxy"); } + public static void setGatewayServiceConfigurationReader( + RxDocumentClientImpl client, + GatewayServiceConfigurationReader gatewayServiceConfigurationReader) { + set(client, gatewayServiceConfigurationReader, "gatewayConfigurationReader"); + } + public static void setServerStoreModel (RxDocumentClientImpl client, RxStoreModel storeModel) { set(client, storeModel, "storeModel"); } diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java index a13bfe770730..2f4a06dd6046 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java @@ -307,7 +307,7 @@ public CosmosAsyncDatabase getDatabase(String id) { @BeforeSuite(groups = {"thinclient", "fast", "long", "direct", "multi-region", "multi-master", "flaky-multi-master", "emulator", "emulator-vnext", "split", "query", "cfp-split", "circuit-breaker-misc-gateway", "circuit-breaker-misc-direct", - "circuit-breaker-read-all-read-many", "fi-multi-master", "fi-customer-workflows", "fi-sm-customer-workflows", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master", "multi-region-strong", "manual-http-network-fault"}, timeOut = SUITE_SETUP_TIMEOUT) + "circuit-breaker-read-all-read-many", "fi-multi-master", "fi-customer-workflows", "fi-sm-customer-workflows", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master", "multi-region-strong", "manual-http-network-fault", "consistency-overrides"}, timeOut = SUITE_SETUP_TIMEOUT) public void beforeSuite() { logger.info("beforeSuite Started"); @@ -358,7 +358,7 @@ private static DocumentCollection getInternalDocumentCollection(CosmosAsyncConta @AfterSuite(groups = {"thinclient", "fast", "long", "direct", "multi-region", "multi-master", "flaky-multi-master", "emulator", "split", "query", "cfp-split", "circuit-breaker-misc-gateway", "circuit-breaker-misc-direct", - "circuit-breaker-read-all-read-many", "fi-multi-master", "fi-customer-workflows", "fi-sm-customer-workflows", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master", "multi-region-strong", "manual-http-network-fault"}, timeOut = SUITE_SHUTDOWN_TIMEOUT) + "circuit-breaker-read-all-read-many", "fi-multi-master", "fi-customer-workflows", "fi-sm-customer-workflows", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master", "multi-region-strong", "manual-http-network-fault", "consistency-overrides"}, timeOut = SUITE_SHUTDOWN_TIMEOUT) public void afterSuite() { logger.info("afterSuite Started"); diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/resources/consistency-overrides-testng.xml b/sdk/cosmos/azure-cosmos-tests/src/test/resources/consistency-overrides-testng.xml new file mode 100644 index 000000000000..5d0a5dc59100 --- /dev/null +++ b/sdk/cosmos/azure-cosmos-tests/src/test/resources/consistency-overrides-testng.xml @@ -0,0 +1,38 @@ + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index eee4f87437f9..25b8545acfff 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -7,6 +7,7 @@ #### Breaking Changes #### Bugs Fixed +* Unified request-level consistency override behavior across transports: invalid attempts to upgrade the request consistency level above the account default are now silently ignored instead of returning `BadRequest` in some gateway paths. - See PR [49606](https://github.com/Azure/azure-sdk-for-java/pull/49606). #### Other Changes * Reduced memory footprint of deserialized `PartitionKeyRange` instances by stripping unused fields in the `PartitionKeyRange(ObjectNode)` constructor - See PR [49513](https://github.com/Azure/azure-sdk-for-java/pull/49513). diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosAsyncClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosAsyncClient.java index fe2b36b766e1..6dbf7c8698d1 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosAsyncClient.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosAsyncClient.java @@ -20,6 +20,7 @@ import com.azure.cosmos.implementation.RequestOptions; import com.azure.cosmos.implementation.ResourceType; import com.azure.cosmos.implementation.Strings; +import com.azure.cosmos.implementation.Utils; import com.azure.cosmos.implementation.WriteRetryPolicy; import com.azure.cosmos.implementation.clienttelemetry.ClientMetricsDiagnosticsHandler; import com.azure.cosmos.implementation.clienttelemetry.ClientTelemetry; @@ -750,11 +751,13 @@ ConsistencyLevel getEffectiveConsistencyLevel( return this.accountConsistencyLevel; } - if (desiredConsistencyLevelOfOperation != null) { + if (desiredConsistencyLevelOfOperation != null + && !Utils.isConsistencyLevelUpgrade(this.accountConsistencyLevel, desiredConsistencyLevelOfOperation)) { return desiredConsistencyLevelOfOperation; } - if (this.desiredConsistencyLevel != null) { + if (this.desiredConsistencyLevel != null + && !Utils.isConsistencyLevelUpgrade(this.accountConsistencyLevel, this.desiredConsistencyLevel)) { return desiredConsistencyLevel; } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java index d4f99c183c24..c3f01ce0afab 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java @@ -2036,9 +2036,7 @@ private Map getRequestHeaders(RequestOptions options, ResourceTy // account's default consistency level in Compute Gateway will result in a 400 Bad Request // even when it is done for resource types / operations where this header should simply be ignored // making the change here to restrict adding the header to when it is relevant. - if ((operationType.isReadOnlyOperation() || operationType == OperationType.Batch) && (resourceType.isMasterResource() || resourceType == ResourceType.Document)) { - headers.put(HttpConstants.HttpHeaders.CONSISTENCY_LEVEL, consistencyLevel.toString()); - } + putConsistencyLevelHeaderIfSupported(headers, consistencyLevel, resourceType, operationType); } if (readConsistencyStrategy != null @@ -2062,6 +2060,8 @@ private Map getRequestHeaders(RequestOptions options, ResourceTy if (!this.contentResponseOnWriteEnabled && resourceType.equals(ResourceType.Document) && operationType.isWriteOperation()) { headers.put(HttpConstants.HttpHeaders.PREFER, HttpConstants.HeaderValues.PREFER_RETURN_MINIMAL); } + + removeUnsupportedConsistencyLevelHeader(headers); return headers; } @@ -2110,9 +2110,11 @@ private Map getRequestHeaders(RequestOptions options, ResourceTy && !headers.containsKey(HttpConstants.HttpHeaders.READ_CONSISTENCY_STRATEGY)) { // Only set ConsistencyLevel when ReadConsistencyStrategy is NOT already present. // readConsistencyStrategy takes precedence — setting both causes gateway rejection. - headers.put(HttpConstants.HttpHeaders.CONSISTENCY_LEVEL, options.getConsistencyLevel().toString()); + putConsistencyLevelHeaderIfSupported(headers, options.getConsistencyLevel(), resourceType, operationType); } + removeUnsupportedConsistencyLevelHeader(headers); + if (options.getIndexingDirective() != null) { headers.put(HttpConstants.HttpHeaders.INDEXING_DIRECTIVE, options.getIndexingDirective().toString()); } @@ -2196,6 +2198,39 @@ private Map getRequestHeaders(RequestOptions options, ResourceTy return headers; } + private void putConsistencyLevelHeaderIfSupported( + Map headers, + ConsistencyLevel requestedConsistencyLevel, + ResourceType resourceType, + OperationType operationType) { + + if (isConsistencyLevelHeaderApplicable(resourceType, operationType) + && !isUnsupportedConsistencyLevelUpgrade(requestedConsistencyLevel)) { + headers.put(HttpConstants.HttpHeaders.CONSISTENCY_LEVEL, requestedConsistencyLevel.toString()); + } + } + + private boolean isConsistencyLevelHeaderApplicable(ResourceType resourceType, OperationType operationType) { + return (operationType.isReadOnlyOperation() || operationType == OperationType.Batch) + && (resourceType.isMasterResource() || resourceType == ResourceType.Document); + } + + private void removeUnsupportedConsistencyLevelHeader(Map headers) { + String requestedConsistencyLevel = headers.get(HttpConstants.HttpHeaders.CONSISTENCY_LEVEL); + if (Strings.isNullOrEmpty(requestedConsistencyLevel)) { + return; + } + + ConsistencyLevel consistencyLevelFromHeader = BridgeInternal.fromServiceSerializedFormat(requestedConsistencyLevel); + if (consistencyLevelFromHeader != null && isUnsupportedConsistencyLevelUpgrade(consistencyLevelFromHeader)) { + headers.remove(HttpConstants.HttpHeaders.CONSISTENCY_LEVEL); + } + } + + private boolean isUnsupportedConsistencyLevelUpgrade(ConsistencyLevel requestedConsistencyLevel) { + return Utils.isConsistencyLevelUpgrade(this.getDefaultConsistencyLevelOfAccount(), requestedConsistencyLevel); + } + public IRetryPolicyFactory getResetSessionTokenRetryPolicy() { return this.resetSessionTokenRetryPolicy; } @@ -2268,6 +2303,7 @@ private void addPartitionKeyInformation(RxDocumentServiceRequest request, ); SerializationDiagnosticsContext serializationDiagnosticsContext = BridgeInternal.getSerializationDiagnosticsContext(request.requestContext.cosmosDiagnostics); + if (serializationDiagnosticsContext != null) { serializationDiagnosticsContext.addSerializationDiagnostics(serializationDiagnostics); } else if (crossRegionAvailabilityContextForRequest != null) { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/Utils.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/Utils.java index 1e57ff4bedca..a387e72a64ce 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/Utils.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/Utils.java @@ -442,6 +442,32 @@ public static boolean isValidConsistency(ConsistencyLevel backendConsistency, } } + public static boolean isConsistencyLevelUpgrade(ConsistencyLevel backendConsistency, + ConsistencyLevel desiredConsistency) { + if (backendConsistency == null || desiredConsistency == null) { + return false; + } + + return getConsistencyLevelRank(desiredConsistency) > getConsistencyLevelRank(backendConsistency); + } + + private static int getConsistencyLevelRank(ConsistencyLevel consistencyLevel) { + switch (consistencyLevel) { + case STRONG: + return 5; + case BOUNDED_STALENESS: + return 4; + case SESSION: + return 3; + case CONSISTENT_PREFIX: + return 2; + case EVENTUAL: + return 1; + default: + throw new IllegalArgumentException("consistencyLevel"); + } + } + public static String getUserAgent() { return getUserAgent(HttpConstants.Versions.SDK_NAME, HttpConstants.Versions.getSdkVersion()); } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/query/DocumentQueryExecutionContextBase.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/query/DocumentQueryExecutionContextBase.java index 4e8d26112316..bc4f6d22e3ec 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/query/DocumentQueryExecutionContextBase.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/query/DocumentQueryExecutionContextBase.java @@ -325,8 +325,11 @@ public Map createCommonHeadersAsync(CosmosQueryRequestOptions co this.client.getReadConsistencyStrategy() == ReadConsistencyStrategy.DEFAULT; } - if (consistencyLevelOverrideApplicable && this.client.getConsistencyLevel() != null) { - requestHeaders.put(HttpConstants.HttpHeaders.CONSISTENCY_LEVEL, this.client.getConsistencyLevel().toString()); + ConsistencyLevel clientConsistencyLevel = this.client.getConsistencyLevel(); + if (consistencyLevelOverrideApplicable + && clientConsistencyLevel != null + && !Utils.isConsistencyLevelUpgrade(this.client.getDefaultConsistencyLevelAsync(), clientConsistencyLevel)) { + requestHeaders.put(HttpConstants.HttpHeaders.CONSISTENCY_LEVEL, clientConsistencyLevel.toString()); } return requestHeaders; diff --git a/sdk/cosmos/live-platform-matrix.json b/sdk/cosmos/live-platform-matrix.json index e5771c249c1c..baa999d6f14a 100644 --- a/sdk/cosmos/live-platform-matrix.json +++ b/sdk/cosmos/live-platform-matrix.json @@ -14,6 +14,7 @@ "-Pcircuit-breaker-read-all-read-many": "CircuitBreakerReadAllAndReadMany", "-Pmulti-region": "MultiRegion", "-Pmulti-region-strong": "MultiRegionStrong", + "-Pconsistency-overrides": "ConsistencyOverrides", "-Plong": "Long", "-DargLine=\"-Dazure.cosmos.directModeProtocol=Tcp\"": "TCP", "Session": "", @@ -51,7 +52,7 @@ "DESIRED_CONSISTENCIES": "[\"Session\"]", "ACCOUNT_CONSISTENCY": "Session", "PROTOCOLS": "[\"Tcp\"]", - "ProfileFlag": [ "-Pcfp-split", "-Psplit", "-Pquery", "-Pfast", "-Pdirect" ], + "ProfileFlag": [ "-Pcfp-split", "-Psplit", "-Pquery", "-Pfast", "-Pdirect", "-Pconsistency-overrides" ], "ArmTemplateParameters": "@{ enableMultipleWriteLocations = $false; defaultConsistencyLevel = 'Session' }", "AdditionalArgs": [ "-DargLine=\"-Dio.netty.handler.ssl.noOpenSsl=true\"" diff --git a/sdk/cosmos/live-thinclient-platform-matrix.json b/sdk/cosmos/live-thinclient-platform-matrix.json index a36eef8a7e7b..9c90b7190ed4 100644 --- a/sdk/cosmos/live-thinclient-platform-matrix.json +++ b/sdk/cosmos/live-thinclient-platform-matrix.json @@ -1,6 +1,7 @@ { "displayNames": { "-Pthinclient": "ThinClient", + "-Pconsistency-overrides": "ConsistencyOverrides", "-Pquery": "Query", "Session": "", "ubuntu": "", @@ -16,7 +17,7 @@ } }, "PROTOCOLS": "[\"Tcp\"]", - "ProfileFlag": [ "-Pthinclient", "-Pquery" ], + "ProfileFlag": [ "-Pthinclient", "-Pquery", "-Pconsistency-overrides" ], "Agent": { "ubuntu": { "OSVmImage": "env:LINUXVMIMAGE", "Pool": "env:LINUXPOOL" } }