diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/pom.xml b/sdk/cosmos/azure-cosmos-kafka-connect/pom.xml index 1693af8dc8fb..c40222186043 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/pom.xml +++ b/sdk/cosmos/azure-cosmos-kafka-connect/pom.xml @@ -63,6 +63,7 @@ Licensed under the MIT License. --add-opens com.azure.cosmos/com.azure.cosmos.implementation.caches=ALL-UNNAMED --add-opens com.azure.cosmos/com.azure.cosmos.implementation.caches=com.azure.cosmos.kafka.connect --add-opens com.azure.cosmos/com.azure.cosmos.implementation.faultinjection=ALL-UNNAMED + --add-opens com.azure.cosmos/com.azure.cosmos.implementation.interceptor=ALL-UNNAMED --add-opens com.azure.cosmos/com.azure.cosmos.implementation.guava25.base=ALL-UNNAMED --add-opens com.azure.cosmos/com.azure.cosmos.implementation.routing=ALL-UNNAMED --add-opens com.azure.cosmos.kafka.connect/com.azure.cosmos.kafka.connect=ALL-UNNAMED diff --git a/sdk/cosmos/azure-cosmos-test/CHANGELOG.md b/sdk/cosmos/azure-cosmos-test/CHANGELOG.md index 47ff60e62d95..a2e41ee8b5a1 100644 --- a/sdk/cosmos/azure-cosmos-test/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-test/CHANGELOG.md @@ -3,12 +3,14 @@ ### 1.0.0-beta.16 (Unreleased) #### Features Added +* Added support for `FaultInjectionOperationType.HEAD_COLLECTION` - See [PR 47231](https://github.com/Azure/azure-sdk-for-java/pull/47231) #### Breaking Changes #### Bugs Fixed #### Other Changes +* Added support for `CosmosTransportClientInterceptor` which allow to modify the store response on direct layer. - See [PR 47231](https://github.com/Azure/azure-sdk-for-java/pull/47231) ### 1.0.0-beta.15 (2025-10-21) diff --git a/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/faultinjection/FaultInjectionOperationType.java b/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/faultinjection/FaultInjectionOperationType.java index 1a1427a91fd9..5b36726e3a9a 100644 --- a/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/faultinjection/FaultInjectionOperationType.java +++ b/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/faultinjection/FaultInjectionOperationType.java @@ -62,5 +62,9 @@ public enum FaultInjectionOperationType { /** * Read change feed items */ - READ_FEED_ITEM + READ_FEED_ITEM, + /** + * Head collection request - barrier request for document operation + */ + HEAD_COLLECTION } diff --git a/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/faultinjection/FaultInjectionRuleProcessor.java b/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/faultinjection/FaultInjectionRuleProcessor.java index bdcbd502c7a0..7652c0950fb5 100644 --- a/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/faultinjection/FaultInjectionRuleProcessor.java +++ b/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/faultinjection/FaultInjectionRuleProcessor.java @@ -351,6 +351,8 @@ private OperationType getEffectiveOperationType(FaultInjectionOperationType faul return OperationType.ReadFeed; case METADATA_REQUEST_ADDRESS_REFRESH: // address refresh can happen for any operations: read, write, query etc return null; + case HEAD_COLLECTION: + return OperationType.Head; default: throw new IllegalStateException("FaultInjectionOperationType " + faultInjectionOperationType + " is not supported"); } @@ -374,6 +376,7 @@ private ResourceType getEffectiveResourceType(FaultInjectionOperationType faultI case METADATA_REQUEST_QUERY_PLAN: return ResourceType.Document; case METADATA_REQUEST_CONTAINER: + case HEAD_COLLECTION: return ResourceType.DocumentCollection; case METADATA_REQUEST_DATABASE_ACCOUNT: return ResourceType.DatabaseAccount; diff --git a/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/interceptor/CosmosInterceptorHelper.java b/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/interceptor/CosmosInterceptorHelper.java new file mode 100644 index 000000000000..2a058ab1bf16 --- /dev/null +++ b/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/interceptor/CosmosInterceptorHelper.java @@ -0,0 +1,24 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.test.implementation.interceptor; + +import com.azure.cosmos.CosmosAsyncClient; +import com.azure.cosmos.implementation.ImplementationBridgeHelpers; +import com.azure.cosmos.implementation.RxDocumentServiceRequest; +import com.azure.cosmos.implementation.directconnectivity.StoreResponse; + +import java.util.function.BiFunction; + +public class CosmosInterceptorHelper { + public static void registerTransportClientInterceptor( + CosmosAsyncClient client, + BiFunction storeResponseInterceptor) { + + CosmosTransportClientInterceptor transportClientInterceptor = new CosmosTransportClientInterceptor(storeResponseInterceptor); + ImplementationBridgeHelpers + .CosmosAsyncClientHelper + .getCosmosAsyncClientAccessor() + .registerTransportClientInterceptor(client, transportClientInterceptor); + } +} diff --git a/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/interceptor/CosmosTransportClientInterceptor.java b/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/interceptor/CosmosTransportClientInterceptor.java new file mode 100644 index 000000000000..ca38c0ca2048 --- /dev/null +++ b/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/interceptor/CosmosTransportClientInterceptor.java @@ -0,0 +1,28 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.test.implementation.interceptor; + +import com.azure.cosmos.implementation.RxDocumentServiceRequest; +import com.azure.cosmos.implementation.directconnectivity.StoreResponse; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; + +import java.util.function.BiFunction; + +import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkNotNull; + +public class CosmosTransportClientInterceptor implements ITransportClientInterceptor { + + private final BiFunction storeResponseInterceptor; + public CosmosTransportClientInterceptor( + BiFunction storeResponseInterceptor) { + + checkNotNull(storeResponseInterceptor, "Argument 'storeResponseInterceptor' must not be null."); + this.storeResponseInterceptor = storeResponseInterceptor; + } + + @Override + public BiFunction getStoreResponseInterceptor() { + return this.storeResponseInterceptor; + } +} diff --git a/sdk/cosmos/azure-cosmos-tests/pom.xml b/sdk/cosmos/azure-cosmos-tests/pom.xml index d02ae9502bdb..13536af9918a 100644 --- a/sdk/cosmos/azure-cosmos-tests/pom.xml +++ b/sdk/cosmos/azure-cosmos-tests/pom.xml @@ -789,5 +789,27 @@ Licensed under the MIT License. + + + fault-injection-barrier + + fault-injection-barrier + + + + + org.apache.maven.plugins + maven-failsafe-plugin + 3.5.3 + + + src/test/resources/fault-injection-barrier-testng.xml + + + + + + + diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/faultinjection/FaultInjectionServerErrorRuleOnDirectTests.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/faultinjection/FaultInjectionServerErrorRuleOnDirectTests.java index d46cb84141a4..58998a9ab6e4 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/faultinjection/FaultInjectionServerErrorRuleOnDirectTests.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/faultinjection/FaultInjectionServerErrorRuleOnDirectTests.java @@ -34,10 +34,12 @@ import com.azure.cosmos.test.faultinjection.FaultInjectionRule; import com.azure.cosmos.test.faultinjection.FaultInjectionRuleBuilder; import com.azure.cosmos.test.faultinjection.FaultInjectionServerErrorType; +import com.azure.cosmos.test.implementation.interceptor.CosmosInterceptorHelper; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ArrayNode; 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; @@ -69,6 +71,7 @@ public class FaultInjectionServerErrorRuleOnDirectTests extends FaultInjectionTe private CosmosAsyncClient clientWithoutPreferredRegions; private CosmosAsyncContainer cosmosAsyncContainer; + private DatabaseAccount databaseAccount; private List accountLevelReadRegions; private List accountLevelWriteRegions; private Map readRegionMap; @@ -81,7 +84,7 @@ public FaultInjectionServerErrorRuleOnDirectTests(CosmosClientBuilder clientBuil this.subscriberValidationTimeout = TIMEOUT; } - @BeforeClass(groups = {"multi-region", "long", "fast", "fi-multi-master"}, timeOut = TIMEOUT) + @BeforeClass(groups = {"multi-region", "long", "fast", "fi-multi-master", "fault-injection-barrier"}, timeOut = TIMEOUT) public void beforeClass() { clientWithoutPreferredRegions = getClientBuilder() .preferredRegions(new ArrayList<>()) @@ -90,7 +93,7 @@ public void beforeClass() { AsyncDocumentClient asyncDocumentClient = BridgeInternal.getContextClient(clientWithoutPreferredRegions); GlobalEndpointManager globalEndpointManager = asyncDocumentClient.getGlobalEndpointManager(); - DatabaseAccount databaseAccount = globalEndpointManager.getLatestDatabaseAccount(); + this.databaseAccount = globalEndpointManager.getLatestDatabaseAccount(); this.cosmosAsyncContainer = getSharedMultiPartitionCosmosContainerWithIdAsPartitionKey(clientWithoutPreferredRegions); AccountLevelLocationContext accountLevelReadableLocationContext @@ -196,6 +199,41 @@ public static Object[] preferredRegionsConfigProvider() { return new Object[] {false, true}; } + @DataProvider(name = "barrierRequestServerErrorResponseProvider") + public static Object[][] barrierRequestServerErrorResponseProvider() { + // OperationType, FaultInjectionErrorType, ErrorStatusCode, ErrorSubStatusCode + return new Object[][] { + // only include exceptions which can be applied by operation type + { OperationType.Create, FaultInjectionServerErrorType.LEASE_NOT_FOUND, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.LEASE_NOT_FOUND }, + { OperationType.Create, FaultInjectionServerErrorType.INTERNAL_SERVER_ERROR, HttpConstants.StatusCodes.INTERNAL_SERVER_ERROR, HttpConstants.SubStatusCodes.UNKNOWN }, + { OperationType.Create, FaultInjectionServerErrorType.RETRY_WITH, HttpConstants.StatusCodes.RETRY_WITH, HttpConstants.SubStatusCodes.UNKNOWN }, + { OperationType.Create, FaultInjectionServerErrorType.TOO_MANY_REQUEST, HttpConstants.StatusCodes.TOO_MANY_REQUESTS, HttpConstants.SubStatusCodes.USER_REQUEST_RATE_TOO_LARGE }, + { OperationType.Create, FaultInjectionServerErrorType.TIMEOUT, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.SERVER_GENERATED_408 }, + { OperationType.Create, FaultInjectionServerErrorType.PARTITION_IS_MIGRATING, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.COMPLETING_PARTITION_MIGRATION }, + { OperationType.Create, FaultInjectionServerErrorType.PARTITION_IS_SPLITTING, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.COMPLETING_SPLIT_OR_MERGE }, + { OperationType.Create, FaultInjectionServerErrorType.SERVICE_UNAVAILABLE, HttpConstants.StatusCodes.SERVICE_UNAVAILABLE, HttpConstants.SubStatusCodes.SERVER_GENERATED_503 }, + { OperationType.Create, FaultInjectionServerErrorType.NAME_CACHE_IS_STALE, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.NAME_CACHE_IS_STALE }, + { OperationType.Read, FaultInjectionServerErrorType.LEASE_NOT_FOUND, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.LEASE_NOT_FOUND }, + { OperationType.Read, FaultInjectionServerErrorType.INTERNAL_SERVER_ERROR, HttpConstants.StatusCodes.INTERNAL_SERVER_ERROR, HttpConstants.SubStatusCodes.UNKNOWN }, + { OperationType.Read, FaultInjectionServerErrorType.RETRY_WITH, HttpConstants.StatusCodes.RETRY_WITH, HttpConstants.SubStatusCodes.UNKNOWN }, + { OperationType.Read, FaultInjectionServerErrorType.TOO_MANY_REQUEST, HttpConstants.StatusCodes.TOO_MANY_REQUESTS, HttpConstants.SubStatusCodes.USER_REQUEST_RATE_TOO_LARGE }, + { OperationType.Read, FaultInjectionServerErrorType.TIMEOUT, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.SERVER_GENERATED_408 }, + { OperationType.Read, FaultInjectionServerErrorType.PARTITION_IS_MIGRATING, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.COMPLETING_PARTITION_MIGRATION }, + { OperationType.Read, FaultInjectionServerErrorType.PARTITION_IS_SPLITTING, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.COMPLETING_SPLIT_OR_MERGE }, + { OperationType.Read, FaultInjectionServerErrorType.SERVICE_UNAVAILABLE, HttpConstants.StatusCodes.SERVICE_UNAVAILABLE, HttpConstants.SubStatusCodes.SERVER_GENERATED_503 }, + { OperationType.Read, FaultInjectionServerErrorType.NAME_CACHE_IS_STALE, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.NAME_CACHE_IS_STALE }, + { OperationType.Query, FaultInjectionServerErrorType.LEASE_NOT_FOUND, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.LEASE_NOT_FOUND }, + { OperationType.Query, FaultInjectionServerErrorType.INTERNAL_SERVER_ERROR, HttpConstants.StatusCodes.INTERNAL_SERVER_ERROR, HttpConstants.SubStatusCodes.UNKNOWN }, + { OperationType.Query, FaultInjectionServerErrorType.RETRY_WITH, HttpConstants.StatusCodes.RETRY_WITH, HttpConstants.SubStatusCodes.UNKNOWN }, + { OperationType.Query, FaultInjectionServerErrorType.TOO_MANY_REQUEST, HttpConstants.StatusCodes.TOO_MANY_REQUESTS, HttpConstants.SubStatusCodes.USER_REQUEST_RATE_TOO_LARGE }, + { OperationType.Query, FaultInjectionServerErrorType.TIMEOUT, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.SERVER_GENERATED_408 }, + { OperationType.Query, FaultInjectionServerErrorType.PARTITION_IS_MIGRATING, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.COMPLETING_PARTITION_MIGRATION }, + { OperationType.Query, FaultInjectionServerErrorType.PARTITION_IS_SPLITTING, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.COMPLETING_SPLIT_OR_MERGE }, + { OperationType.Query, FaultInjectionServerErrorType.SERVICE_UNAVAILABLE, HttpConstants.StatusCodes.SERVICE_UNAVAILABLE, HttpConstants.SubStatusCodes.SERVER_GENERATED_503 }, + { OperationType.Query, FaultInjectionServerErrorType.NAME_CACHE_IS_STALE, HttpConstants.StatusCodes.GONE, HttpConstants.SubStatusCodes.NAME_CACHE_IS_STALE } + }; + } + @Test(groups = {"multi-region", "long"}, dataProvider = "operationTypeProvider", timeOut = TIMEOUT) public void faultInjectionServerErrorRuleTests_OperationType(OperationType operationType) throws JsonProcessingException { // Test for SERVER_GONE, the operation type will be ignored after getting the addresses @@ -1019,7 +1057,7 @@ public void faultInjectionServerErrorRuleTests_HitLimit() throws JsonProcessingE } } - @AfterClass(groups = {"multi-region", "long", "fast", "fi-multi-master"}, timeOut = SHUTDOWN_TIMEOUT, alwaysRun = true) + @AfterClass(groups = {"multi-region", "long", "fast", "fi-multi-master", "fault-injection-barrier"}, timeOut = SHUTDOWN_TIMEOUT, alwaysRun = true) public void afterClass() { safeClose(clientWithoutPreferredRegions); } @@ -1437,6 +1475,85 @@ public void faultInjectionInjectTcpResponseDelay() throws JsonProcessingExceptio } } + @Test(groups = {"fault-injection-barrier"}, dataProvider = "barrierRequestServerErrorResponseProvider", timeOut = 2 * TIMEOUT) + public void faultInjection_serverError_barrierRequest( + OperationType operationType, + FaultInjectionServerErrorType serverErrorType, + int statusCode, + int subStatusCode) throws JsonProcessingException { + + // Test to verify server error type can be injected to barrier requests + + // for barrier request flow, only test on strong consistency + if (this.databaseAccount.getConsistencyPolicy().getDefaultConsistencyLevel() != ConsistencyLevel.STRONG) { + throw new SkipException( + String.format( + "Test is not applicable to %s consistency level!", + this.databaseAccount.getConsistencyPolicy().getDefaultConsistencyLevel())); + } + + CosmosAsyncClient newClient = null; + String faultInjectionRuleId = "barrier-" + serverErrorType + "-" + UUID.randomUUID(); + FaultInjectionRule faultInjectionRule = + new FaultInjectionRuleBuilder(faultInjectionRuleId) + .condition( + new FaultInjectionConditionBuilder() + .operationType(FaultInjectionOperationType.HEAD_COLLECTION) + .build() + ) + .result( + FaultInjectionResultBuilders + .getResultBuilder(serverErrorType) + .times(2) + .build() + ) + .duration(Duration.ofMinutes(5)) + .build(); + + try { + newClient = new CosmosClientBuilder() + .endpoint(TestConfigurations.HOST) + .key(TestConfigurations.MASTER_KEY) + .contentResponseOnWriteEnabled(true) + .buildAsyncClient(); + + CosmosAsyncContainer container = + newClient + .getDatabase(cosmosAsyncContainer.getDatabase().getId()) + .getContainer(cosmosAsyncContainer.getId()); + + TestObject testItem = TestObject.create(); + container.createItem(testItem).block(); + + CosmosFaultInjectionHelper.configureFaultInjectionRules(container, Arrays.asList(faultInjectionRule)).block(); + + // in order to trigger barrier request, we will need to also modify the store response of the original read/write operation so that GCLSN < LSN + CosmosInterceptorHelper.registerTransportClientInterceptor( + newClient, + (request, storeResponse) -> { + if (request.getResourceType() == ResourceType.Document && request.getOperationType() == operationType) { + // Decrement so that GCLSN < LSN to simulate the replication lag + logger.info("faultInjection_serverError_barrierRequest reducing gclsn"); + storeResponse.setGCLSN(storeResponse.getLSN() - 2L); + } + return storeResponse; + } + ); + + CosmosDiagnostics cosmosDiagnostics = this.performDocumentOperation(container, operationType, testItem, false); + validateFaultInjectionRuleAppliedForBarrier( + cosmosDiagnostics, + operationType, + statusCode, + subStatusCode, + faultInjectionRule.getId()); + + } finally { + faultInjectionRule.disable(); + safeClose(newClient); + } + } + private void validateFaultInjectionRuleApplied( CosmosDiagnostics cosmosDiagnostics, OperationType operationType, @@ -1445,6 +1562,42 @@ private void validateFaultInjectionRuleApplied( String ruleId, boolean canRetryOnFaultInjectedError) throws JsonProcessingException { + validateFaultInjectionRuleApplied( + cosmosDiagnostics, + operationType, + statusCode, + subStatusCode, + ruleId, + canRetryOnFaultInjectedError, + false); + } + + private void validateFaultInjectionRuleAppliedForBarrier( + CosmosDiagnostics cosmosDiagnostics, + OperationType operationType, + int statusCode, + int subStatusCode, + String ruleId) throws JsonProcessingException { + + validateFaultInjectionRuleApplied( + cosmosDiagnostics, + operationType, + statusCode, + subStatusCode, + ruleId, + true, + true); + } + + private void validateFaultInjectionRuleApplied( + CosmosDiagnostics cosmosDiagnostics, + OperationType operationType, + int statusCode, + int subStatusCode, + String ruleId, + boolean canRetryOnFaultInjectedError, + boolean validateForBarrier) throws JsonProcessingException { + List clientSideRequestStatisticsNodes = new ArrayList<>(); assertThat(cosmosDiagnostics.getDiagnosticsContext()).isNotNull(); @@ -1462,8 +1615,10 @@ private void validateFaultInjectionRuleApplied( } List responseStatisticsNodes = new ArrayList<>(); + + String diagnosticsNodeName = validateForBarrier ? "supplementalResponseStatisticsList" : "responseStatisticsList"; for (ObjectNode diagnosticNode : clientSideRequestStatisticsNodes) { - JsonNode responseStatisticsList = diagnosticNode.get("responseStatisticsList"); + JsonNode responseStatisticsList = diagnosticNode.get(diagnosticsNodeName); assertThat(responseStatisticsList.isArray()).isTrue(); for (JsonNode responseStatisticsNode : responseStatisticsList) { 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 d17e872db25e..6f18de03873e 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 @@ -205,7 +205,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", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master"}, timeOut = SUITE_SETUP_TIMEOUT) + "circuit-breaker-read-all-read-many", "fi-multi-master", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master", "fault-injection-barrier"}, timeOut = SUITE_SETUP_TIMEOUT) public void beforeSuite() { logger.info("beforeSuite Started"); @@ -230,7 +230,7 @@ public void parallelizeUnitTests(ITestContext context) { @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", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master"}, timeOut = SUITE_SHUTDOWN_TIMEOUT) + "circuit-breaker-read-all-read-many", "fi-multi-master", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master", "fault-injection-barrier"}, timeOut = SUITE_SHUTDOWN_TIMEOUT) public void afterSuite() { logger.info("afterSuite Started"); diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/resources/fault-injection-barrier-testng.xml b/sdk/cosmos/azure-cosmos-tests/src/test/resources/fault-injection-barrier-testng.xml new file mode 100644 index 000000000000..6444f7176010 --- /dev/null +++ b/sdk/cosmos/azure-cosmos-tests/src/test/resources/fault-injection-barrier-testng.xml @@ -0,0 +1,38 @@ + + + + + + + + + + + + + + + + + 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 680b98a6a915..ec0dd64af008 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 @@ -29,6 +29,7 @@ import com.azure.cosmos.implementation.clienttelemetry.TagName; import com.azure.cosmos.implementation.directconnectivity.rntbd.RntbdMetrics; import com.azure.cosmos.implementation.faultinjection.IFaultInjectorProvider; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.throughputControl.sdk.config.SDKThroughputControlGroupInternal; import com.azure.cosmos.implementation.throughputControl.server.config.ServerThroughputControlGroup; import com.azure.cosmos.models.CosmosAuthorizationTokenResolver; @@ -794,6 +795,10 @@ CosmosItemSerializer getEffectiveItemSerializer(CosmosItemSerializer requestOpti return this.asyncDocumentClient.getEffectiveItemSerializer(requestOptionsItemSerializer); } + void registerTransportClientInterceptor(ITransportClientInterceptor transportClientInterceptor) { + this.asyncDocumentClient.registerTransportClientInterceptor(transportClientInterceptor); + } + boolean isTransportLevelTracingEnabled() { CosmosClientTelemetryConfig effectiveConfig = this.clientTelemetryConfig != null ? @@ -950,6 +955,14 @@ public List getOperationPolicies(CosmosAsyncClient client public CosmosItemSerializer getEffectiveItemSerializer(CosmosAsyncClient client, CosmosItemSerializer requestOptionsItemSerializer) { return client.getEffectiveItemSerializer(requestOptionsItemSerializer); } + + @Override + public void registerTransportClientInterceptor( + CosmosAsyncClient client, + ITransportClientInterceptor transportClientInterceptor) { + + client.registerTransportClientInterceptor(transportClientInterceptor); + } } ); } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/AsyncDocumentClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/AsyncDocumentClient.java index f19ccb503027..d0df0de41b45 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/AsyncDocumentClient.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/AsyncDocumentClient.java @@ -15,6 +15,7 @@ import com.azure.cosmos.implementation.batch.ServerBatchRequest; import com.azure.cosmos.implementation.caches.RxClientCollectionCache; import com.azure.cosmos.implementation.caches.RxPartitionKeyRangeCache; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.perPartitionAutomaticFailover.GlobalPartitionEndpointManagerForPerPartitionAutomaticFailover; import com.azure.cosmos.implementation.perPartitionCircuitBreaker.GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker; import com.azure.cosmos.implementation.clienttelemetry.ClientTelemetry; @@ -1678,5 +1679,7 @@ Flux> readAllDocuments( */ void recordOpenConnectionsAndInitCachesStarted(List cosmosContainerIdentities); - public String getMasterKeyOrResourceToken(); + String getMasterKeyOrResourceToken(); + + void registerTransportClientInterceptor(ITransportClientInterceptor transportClientInterceptor); } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java index 735f37e4accd..d2f5cde47bac 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java @@ -42,6 +42,7 @@ import com.azure.cosmos.implementation.directconnectivity.Uri; import com.azure.cosmos.implementation.directconnectivity.rntbd.RntbdChannelStatistics; import com.azure.cosmos.implementation.faultinjection.IFaultInjectorProvider; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.patch.PatchOperation; import com.azure.cosmos.implementation.routing.PartitionKeyInternal; import com.azure.cosmos.implementation.routing.RegionalRoutingContext; @@ -1470,6 +1471,10 @@ CosmosDiagnosticsThresholds getEffectiveDiagnosticsThresholds( CosmosItemSerializer getEffectiveItemSerializer( CosmosAsyncClient client, CosmosItemSerializer requestOptionsItemSerializer); + + void registerTransportClientInterceptor( + CosmosAsyncClient cosmosAsyncClient, + ITransportClientInterceptor transportClientInterceptor); } } 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 25f02086acc3..bdb2cac11aaa 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 @@ -47,6 +47,7 @@ import com.azure.cosmos.implementation.http.HttpClientConfig; import com.azure.cosmos.implementation.http.HttpHeaders; import com.azure.cosmos.implementation.http.SharedGatewayHttpClient; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.patch.PatchUtil; import com.azure.cosmos.implementation.perPartitionAutomaticFailover.GlobalPartitionEndpointManagerForPerPartitionAutomaticFailover; import com.azure.cosmos.implementation.perPartitionCircuitBreaker.GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker; @@ -6582,6 +6583,11 @@ public String getMasterKeyOrResourceToken() { return this.masterKeyOrResourceToken; } + @Override + public void registerTransportClientInterceptor(ITransportClientInterceptor transportClientInterceptor) { + this.storeModel.registerTransportClientInterceptor(transportClientInterceptor); + } + private static SqlQuerySpec createLogicalPartitionScanQuerySpec( PartitionKey partitionKey, List partitionKeySelectors) { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxGatewayStoreModel.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxGatewayStoreModel.java index 680b591d38c9..b7455ade6906 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxGatewayStoreModel.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxGatewayStoreModel.java @@ -24,6 +24,7 @@ import com.azure.cosmos.implementation.http.HttpResponse; import com.azure.cosmos.implementation.http.HttpTransportSerializer; import com.azure.cosmos.implementation.http.ReactorNettyRequestRecord; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.perPartitionCircuitBreaker.GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker; import com.azure.cosmos.implementation.perPartitionCircuitBreaker.LocationSpecificHealthContext; import com.azure.cosmos.implementation.routing.PartitionKeyInternal; @@ -767,6 +768,11 @@ public void recordOpenConnectionsAndInitCachesStarted(List getDefaultHeaders() { return this.defaultHeaders; } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxStoreModel.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxStoreModel.java index 6629aa860928..58876a54f463 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxStoreModel.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxStoreModel.java @@ -4,6 +4,7 @@ import com.azure.cosmos.CosmosContainerProactiveInitConfig; import com.azure.cosmos.implementation.faultinjection.IFaultInjectorProvider; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.spark.OperationContext; import com.azure.cosmos.implementation.spark.OperationContextAndListenerTuple; import com.azure.cosmos.implementation.spark.OperationListener; @@ -75,4 +76,5 @@ default Mono processMessage(RxDocumentServiceRequest void configureFaultInjectorProvider(IFaultInjectorProvider injectorProvider, Configs configs); void recordOpenConnectionsAndInitCachesCompleted(List cosmosContainerIdentities); void recordOpenConnectionsAndInitCachesStarted(List cosmosContainerIdentities); + void registerTransportClientInterceptor(ITransportClientInterceptor transportClientInterceptor); } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/BarrierRequestHelper.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/BarrierRequestHelper.java index a39546e485cf..55a3534eeec1 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/BarrierRequestHelper.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/BarrierRequestHelper.java @@ -131,6 +131,10 @@ public static Mono createAsync( barrierLsnRequest.requestContext = request.requestContext.clone(); + // barrier requests are associated with the write/read operation + // copy over the fault injection context here + barrierLsnRequest.faultInjectionRequestContext = request.faultInjectionRequestContext; + if (request.getPartitionKeyRangeIdentity() != null) { barrierLsnRequest.routeTo(request.getPartitionKeyRangeIdentity()); } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ReplicatedResourceClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ReplicatedResourceClient.java index 17a57fa2bc09..0e9c454394b2 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ReplicatedResourceClient.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ReplicatedResourceClient.java @@ -18,6 +18,7 @@ import com.azure.cosmos.implementation.ResourceType; import com.azure.cosmos.implementation.RxDocumentServiceRequest; import com.azure.cosmos.implementation.faultinjection.IFaultInjectorProvider; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.throughputControl.ThroughputControlStore; import com.azure.cosmos.models.CosmosContainerIdentity; import reactor.core.publisher.Flux; @@ -170,4 +171,8 @@ public Flux submitOpenConnectionTasksAndInitCaches(CosmosContainerProactiv public void configureFaultInjectorProvider(IFaultInjectorProvider injectorProvider) { this.transportClient.configureFaultInjectorProvider(injectorProvider); } + + public void registerTransportClientInterceptor(ITransportClientInterceptor transportClientInterceptor) { + this.transportClient.registerTransportClientInterceptor(transportClientInterceptor); + } } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ServerStoreModel.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ServerStoreModel.java index c801c8ff3273..07a32cb53fc4 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ServerStoreModel.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ServerStoreModel.java @@ -18,6 +18,7 @@ import com.azure.cosmos.implementation.RxStoreModel; import com.azure.cosmos.implementation.Strings; import com.azure.cosmos.implementation.faultinjection.IFaultInjectorProvider; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.throughputControl.ThroughputControlStore; import com.azure.cosmos.models.CosmosContainerIdentity; import reactor.core.publisher.Flux; @@ -101,4 +102,9 @@ public void recordOpenConnectionsAndInitCachesCompleted(List cosmosContainerIdentities) { this.storeClient.recordOpenConnectionsAndInitCachesStarted(cosmosContainerIdentities); } + + @Override + public void registerTransportClientInterceptor(ITransportClientInterceptor transportClientInterceptor) { + this.storeClient.registerTransportClientInterceptor(transportClientInterceptor); + } } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreClient.java index a01d6cfcac6e..8c1c270bf5cd 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreClient.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreClient.java @@ -29,6 +29,7 @@ import com.azure.cosmos.implementation.Utils; import com.azure.cosmos.implementation.apachecommons.lang.math.NumberUtils; import com.azure.cosmos.implementation.faultinjection.IFaultInjectorProvider; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.throughputControl.ThroughputControlStore; import com.azure.cosmos.models.CosmosContainerIdentity; import org.slf4j.Logger; @@ -155,6 +156,10 @@ public void recordOpenConnectionsAndInitCachesStarted(List transportClientInterceptors; public void enableThroughputControl(ThroughputControlStore throughputControlStore) { this.throughputControlStore = throughputControlStore; } - // Uses requests's ResourceOperation to determine the operation public Mono invokeResourceOperationAsync(Uri physicalAddress, RxDocumentServiceRequest request) { + return invokeResourceOperationInternalAsync(physicalAddress, request) + .map(response -> { + if (this.transportClientInterceptors == null) { + return response; + } else { + // there are interceptors configured, process the store response with the interceptors + StoreResponse storeResponse = response; + for (ITransportClientInterceptor transportClientInterceptor : this.transportClientInterceptors) { + if (transportClientInterceptor.getStoreResponseInterceptor() != null) { + storeResponse = transportClientInterceptor.getStoreResponseInterceptor().apply(request, storeResponse); + } + } + + return storeResponse; + } + }); + } + + // Uses request's ResourceOperation to determine the operation + public Mono invokeResourceOperationInternalAsync(Uri physicalAddress, RxDocumentServiceRequest request) { if (StringUtils.isEmpty(request.requestContext.resourcePhysicalAddress)) { request.requestContext.resourcePhysicalAddress = physicalAddress.toString(); } @@ -57,6 +79,14 @@ protected abstract Mono invokeStoreAsync( public abstract void recordOpenConnectionsAndInitCachesStarted(List cosmosContainerIdentities); + public synchronized void registerTransportClientInterceptor(ITransportClientInterceptor transportClientInterceptor) { + if (this.transportClientInterceptors == null) { + this.transportClientInterceptors = new ArrayList<>(); + } + + this.transportClientInterceptors.add(transportClientInterceptor); + } + private Mono invokeStoreWithThroughputControlAsync(Uri physicalAddress, RxDocumentServiceRequest request) { return this.throughputControlStore.processRequest( request, diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/interceptor/ITransportClientInterceptor.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/interceptor/ITransportClientInterceptor.java new file mode 100644 index 000000000000..bb0d263a8621 --- /dev/null +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/interceptor/ITransportClientInterceptor.java @@ -0,0 +1,13 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.implementation.interceptor; + +import com.azure.cosmos.implementation.RxDocumentServiceRequest; +import com.azure.cosmos.implementation.directconnectivity.StoreResponse; + +import java.util.function.BiFunction; + +public interface ITransportClientInterceptor { + BiFunction getStoreResponseInterceptor(); +} diff --git a/sdk/cosmos/azure-cosmos/src/main/java/module-info.java b/sdk/cosmos/azure-cosmos/src/main/java/module-info.java index 82b7f11e5056..f69228ebe38b 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/module-info.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/module-info.java @@ -82,6 +82,8 @@ opens com.azure.cosmos.implementation.throughputControl.sdk to com.fasterxml.jackson.databind; opens com.azure.cosmos.implementation.throughputControl.sdk.config to com.fasterxml.jackson.databind; + exports com.azure.cosmos.implementation.interceptor to com.azure.cosmos.test; + uses com.azure.cosmos.implementation.guava25.base.PatternCompiler; uses com.azure.core.util.tracing.Tracer; } diff --git a/sdk/cosmos/live-platform-matrix.json b/sdk/cosmos/live-platform-matrix.json index 7f3e3af80997..96581316474d 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", "-Plong": "Long", + "-Pfault-injection-barrier": "Fault-injection-barrier", "-DargLine=\"-Dazure.cosmos.directModeProtocol=Tcp\"": "TCP", "Session": "", "ubuntu": "", @@ -21,7 +22,8 @@ "@{ enableMultipleWriteLocations = $true; defaultConsistencyLevel = 'Session' }": "", "@{ enableMultipleWriteLocations = $false; defaultConsistencyLevel = 'Strong' }": "", "@{ enableMultipleWriteLocations = $false; defaultConsistencyLevel = 'Session'; enablePartitionMerge = $true }": "", - "@{ enableMultipleWriteLocations = $false; defaultConsistencyLevel = 'Session'; enableMultipleRegions = $true; enablePartitionMerge = $true}": "" + "@{ enableMultipleWriteLocations = $false; defaultConsistencyLevel = 'Session'; enableMultipleRegions = $true; enablePartitionMerge = $true}": "", + "@{ enableMultipleWriteLocations = $false; defaultConsistencyLevel = 'Strong'; enableMultipleRegions = $true }": "" }, "include": [ { @@ -176,6 +178,16 @@ "Agent": { "ubuntu": { "OSVmImage": "env:LINUXVMIMAGE", "Pool": "env:LINUXPOOL" } } + }, + { + "DESIRED_CONSISTENCIES": "[\"Strong\"]", + "ACCOUNT_CONSISTENCY": "Strong", + "PROTOCOLS": "[\"Tcp\"]", + "ProfileFlag": [ "-Pfault-injection-barrier" ], + "ArmTemplateParameters": "@{ enableMultipleWriteLocations = $false; defaultConsistencyLevel = 'Strong'; enableMultipleRegions = $true }", + "Agent": { + "ubuntu": { "OSVmImage": "env:LINUXVMIMAGE", "Pool": "env:LINUXPOOL" } + } } ] }