From e220cb37b712266535bf88f55babbbef1f88eb07 Mon Sep 17 00:00:00 2001 From: annie-mac Date: Tue, 11 Nov 2025 11:27:22 -0800 Subject: [PATCH 1/7] add support for head collection request --- .../FaultInjectionOperationType.java | 6 +- .../FaultInjectionRuleProcessor.java | 3 + .../interceptor/CosmosInterceptorHelper.java | 24 +++ .../CosmosTransportClientInterceptor.java | 24 +++ ...InjectionServerErrorRuleOnDirectTests.java | 158 +++++++++++++++++- .../com/azure/cosmos/CosmosAsyncClient.java | 13 ++ .../implementation/AsyncDocumentClient.java | 5 +- .../ImplementationBridgeHelpers.java | 5 + .../implementation/RxDocumentClientImpl.java | 6 + .../implementation/RxGatewayStoreModel.java | 6 + .../cosmos/implementation/RxStoreModel.java | 2 + .../BarrierRequestHelper.java | 4 + .../ReplicatedResourceClient.java | 5 + .../directconnectivity/ServerStoreModel.java | 6 + .../directconnectivity/StoreClient.java | 5 + .../directconnectivity/StoreResponse.java | 18 ++ .../directconnectivity/TransportClient.java | 32 +++- .../ITransportClientInterceptor.java | 13 ++ .../src/main/java/module-info.java | 2 + 19 files changed, 332 insertions(+), 5 deletions(-) create mode 100644 sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/interceptor/CosmosInterceptorHelper.java create mode 100644 sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/interceptor/CosmosTransportClientInterceptor.java create mode 100644 sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/interceptor/ITransportClientInterceptor.java 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..e5822caffd06 --- /dev/null +++ b/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/interceptor/CosmosTransportClientInterceptor.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.implementation.RxDocumentServiceRequest; +import com.azure.cosmos.implementation.directconnectivity.StoreResponse; +import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; + +import java.util.function.BiFunction; + +public class CosmosTransportClientInterceptor implements ITransportClientInterceptor { + + private final BiFunction storeResponseInterceptor; + public CosmosTransportClientInterceptor( + BiFunction storeResponseInterceptor) { + this.storeResponseInterceptor = storeResponseInterceptor; + } + + @Override + public BiFunction getStoreResponseInterceptor() { + return this.storeResponseInterceptor; + } +} 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..61bccadee407 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; @@ -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 @@ -1437,6 +1475,84 @@ public void faultInjectionInjectTcpResponseDelay() throws JsonProcessingExceptio } } + @Test(groups = {"fast"}, 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 + 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 +1561,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 +1614,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/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 ea81761b0411..1bced19b56fd 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; @@ -749,6 +750,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; } From 3f01c0a2effed1b5883048b214c900662c8f9927 Mon Sep 17 00:00:00 2001 From: annie-mac Date: Tue, 11 Nov 2025 11:38:00 -0800 Subject: [PATCH 2/7] update changelog --- sdk/cosmos/azure-cosmos-test/CHANGELOG.md | 2 ++ .../interceptor/CosmosTransportClientInterceptor.java | 4 ++++ sdk/cosmos/azure-cosmos/CHANGELOG.md | 1 + 3 files changed, 7 insertions(+) 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/implementation/interceptor/CosmosTransportClientInterceptor.java b/sdk/cosmos/azure-cosmos-test/src/main/java/com/azure/cosmos/test/implementation/interceptor/CosmosTransportClientInterceptor.java index e5822caffd06..ca38c0ca2048 100644 --- 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 @@ -9,11 +9,15 @@ 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; } diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index 4a0bbb04fbf0..a85f0a75a2bd 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -10,6 +10,7 @@ #### Other Changes * Changed to use incremental change feed to get partition key ranges. - [46810](https://github.com/Azure/azure-sdk-for-java/pull/46810) +* Added support for `ITransportClientInterceptor` which allow to modify the store response on direct layer. - See [PR 47231](https://github.com/Azure/azure-sdk-for-java/pull/47231) ### 4.75.0 (2025-10-21) > [!IMPORTANT] From d51837c8a43604100a89354f9ec3bbd6a92d546c Mon Sep 17 00:00:00 2001 From: annie-mac Date: Tue, 11 Nov 2025 12:26:56 -0800 Subject: [PATCH 3/7] resolve comments --- .../FaultInjectionServerErrorRuleOnDirectTests.java | 2 +- .../cosmos/implementation/directconnectivity/StoreResponse.java | 1 + .../implementation/directconnectivity/TransportClient.java | 2 +- 3 files changed, 3 insertions(+), 2 deletions(-) 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 61bccadee407..d948162b2552 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 @@ -1539,7 +1539,7 @@ public void faultInjection_serverError_barrierRequest( } ); - CosmosDiagnostics cosmosDiagnostics = this.performDocumentOperation(container, operationType, testItem,false); + CosmosDiagnostics cosmosDiagnostics = this.performDocumentOperation(container, operationType, testItem, false); validateFaultInjectionRuleAppliedForBarrier( cosmosDiagnostics, operationType, diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponse.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponse.java index f623a633f042..52681e2ed6fa 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponse.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponse.java @@ -210,6 +210,7 @@ private void setHeaderValue(String headerName, String value) { for (int i = 0; i < responseHeaderNames.length; i++) { if (responseHeaderNames[i].equalsIgnoreCase(headerName)) { responseHeaderValues[i] = value; + break; } } } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java index f88bc598af7b..2498e5807a4a 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java @@ -42,7 +42,7 @@ public Mono invokeResourceOperationAsync(Uri physicalAddress, RxD // there are interceptors configured, process the store response with the interceptors StoreResponse storeResponse = response; for (ITransportClientInterceptor transportClientInterceptor : this.transportClientInterceptors) { - if(transportClientInterceptor.getStoreResponseInterceptor() != null) { + if (transportClientInterceptor.getStoreResponseInterceptor() != null) { storeResponse = transportClientInterceptor.getStoreResponseInterceptor().apply(request, storeResponse); } } From c4c9648b21bcc71626581bb85902c99234009912 Mon Sep 17 00:00:00 2001 From: annie-mac Date: Tue, 11 Nov 2025 12:58:31 -0800 Subject: [PATCH 4/7] resolve comments --- sdk/cosmos/azure-cosmos-kafka-connect/pom.xml | 1 + sdk/cosmos/azure-cosmos/CHANGELOG.md | 1 - 2 files changed, 1 insertion(+), 1 deletion(-) 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/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index a85f0a75a2bd..4a0bbb04fbf0 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -10,7 +10,6 @@ #### Other Changes * Changed to use incremental change feed to get partition key ranges. - [46810](https://github.com/Azure/azure-sdk-for-java/pull/46810) -* Added support for `ITransportClientInterceptor` which allow to modify the store response on direct layer. - See [PR 47231](https://github.com/Azure/azure-sdk-for-java/pull/47231) ### 4.75.0 (2025-10-21) > [!IMPORTANT] From 1bb537b7a283bf96af0e1918c90970472661c728 Mon Sep 17 00:00:00 2001 From: annie-mac Date: Thu, 13 Nov 2025 21:40:50 -0800 Subject: [PATCH 5/7] add log --- .../FaultInjectionServerErrorRuleOnDirectTests.java | 1 + 1 file changed, 1 insertion(+) 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 d948162b2552..bc0bd1f572ce 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 @@ -1540,6 +1540,7 @@ public void faultInjection_serverError_barrierRequest( ); CosmosDiagnostics cosmosDiagnostics = this.performDocumentOperation(container, operationType, testItem, false); + logger.info("faultInjection_serverError_barrierRequest cosmosDiagnostics: " + cosmosDiagnostics.toString()); validateFaultInjectionRuleAppliedForBarrier( cosmosDiagnostics, operationType, From 5f7dcb71ea873cbb019e3c7b069ea2ba92c5ebc7 Mon Sep 17 00:00:00 2001 From: annie-mac Date: Fri, 14 Nov 2025 09:57:53 -0800 Subject: [PATCH 6/7] add extra log --- .../FaultInjectionServerErrorRuleOnDirectTests.java | 2 ++ .../implementation/directconnectivity/TransportClient.java | 6 ++++++ 2 files changed, 8 insertions(+) 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 bc0bd1f572ce..29b69f7929ca 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 @@ -1528,11 +1528,13 @@ public void faultInjection_serverError_barrierRequest( 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 + logger.info("faultInjection_serverError_barrierRequest RegisterTransportClientInterceptor"); 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; diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java index 2498e5807a4a..039f5b6f9b4a 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java @@ -13,6 +13,8 @@ import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.throughputControl.ThroughputControlStore; import com.azure.cosmos.models.CosmosContainerIdentity; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import reactor.core.publisher.Mono; import java.util.ArrayList; @@ -25,6 +27,7 @@ // signature for backwards compatibility purposes. @SuppressWarnings("try") public abstract class TransportClient implements AutoCloseable { + private static final Logger logger = LoggerFactory.getLogger(TransportClient.class); private final boolean switchOffIOThreadForResponse = Configs.shouldSwitchOffIOThreadForResponse(); private ThroughputControlStore throughputControlStore; private List transportClientInterceptors; @@ -37,8 +40,10 @@ public Mono invokeResourceOperationAsync(Uri physicalAddress, RxD return invokeResourceOperationInternalAsync(physicalAddress, request) .map(response -> { if (this.transportClientInterceptors == null) { + logger.info("invokeResourceOperationAsync without interceptor"); return response; } else { + logger.info("invokeResourceOperationAsync with interceptor"); // there are interceptors configured, process the store response with the interceptors StoreResponse storeResponse = response; for (ITransportClientInterceptor transportClientInterceptor : this.transportClientInterceptors) { @@ -84,6 +89,7 @@ public synchronized void registerTransportClientInterceptor(ITransportClientInte this.transportClientInterceptors = new ArrayList<>(); } + logger.info("registerTransportClientInterceptor in Transport client"); this.transportClientInterceptors.add(transportClientInterceptor); } From f73157128a3e6d9bfc2d930a528d86c561572ffc Mon Sep 17 00:00:00 2001 From: annie-mac Date: Mon, 17 Nov 2025 16:17:31 -0800 Subject: [PATCH 7/7] add fault-injection-barrier test suite --- sdk/cosmos/azure-cosmos-tests/pom.xml | 22 +++++++++++ ...InjectionServerErrorRuleOnDirectTests.java | 8 ++-- .../com/azure/cosmos/rx/TestSuiteBase.java | 4 +- .../fault-injection-barrier-testng.xml | 38 +++++++++++++++++++ .../directconnectivity/TransportClient.java | 6 --- sdk/cosmos/live-platform-matrix.json | 14 ++++++- 6 files changed, 78 insertions(+), 14 deletions(-) create mode 100644 sdk/cosmos/azure-cosmos-tests/src/test/resources/fault-injection-barrier-testng.xml 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 29b69f7929ca..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 @@ -84,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<>()) @@ -1057,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); } @@ -1475,7 +1475,7 @@ public void faultInjectionInjectTcpResponseDelay() throws JsonProcessingExceptio } } - @Test(groups = {"fast"}, dataProvider = "barrierRequestServerErrorResponseProvider", timeOut = 2 * TIMEOUT) + @Test(groups = {"fault-injection-barrier"}, dataProvider = "barrierRequestServerErrorResponseProvider", timeOut = 2 * TIMEOUT) public void faultInjection_serverError_barrierRequest( OperationType operationType, FaultInjectionServerErrorType serverErrorType, @@ -1528,7 +1528,6 @@ public void faultInjection_serverError_barrierRequest( 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 - logger.info("faultInjection_serverError_barrierRequest RegisterTransportClientInterceptor"); CosmosInterceptorHelper.registerTransportClientInterceptor( newClient, (request, storeResponse) -> { @@ -1542,7 +1541,6 @@ public void faultInjection_serverError_barrierRequest( ); CosmosDiagnostics cosmosDiagnostics = this.performDocumentOperation(container, operationType, testItem, false); - logger.info("faultInjection_serverError_barrierRequest cosmosDiagnostics: " + cosmosDiagnostics.toString()); validateFaultInjectionRuleAppliedForBarrier( cosmosDiagnostics, operationType, 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/implementation/directconnectivity/TransportClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java index 039f5b6f9b4a..2498e5807a4a 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/TransportClient.java @@ -13,8 +13,6 @@ import com.azure.cosmos.implementation.interceptor.ITransportClientInterceptor; import com.azure.cosmos.implementation.throughputControl.ThroughputControlStore; import com.azure.cosmos.models.CosmosContainerIdentity; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import reactor.core.publisher.Mono; import java.util.ArrayList; @@ -27,7 +25,6 @@ // signature for backwards compatibility purposes. @SuppressWarnings("try") public abstract class TransportClient implements AutoCloseable { - private static final Logger logger = LoggerFactory.getLogger(TransportClient.class); private final boolean switchOffIOThreadForResponse = Configs.shouldSwitchOffIOThreadForResponse(); private ThroughputControlStore throughputControlStore; private List transportClientInterceptors; @@ -40,10 +37,8 @@ public Mono invokeResourceOperationAsync(Uri physicalAddress, RxD return invokeResourceOperationInternalAsync(physicalAddress, request) .map(response -> { if (this.transportClientInterceptors == null) { - logger.info("invokeResourceOperationAsync without interceptor"); return response; } else { - logger.info("invokeResourceOperationAsync with interceptor"); // there are interceptors configured, process the store response with the interceptors StoreResponse storeResponse = response; for (ITransportClientInterceptor transportClientInterceptor : this.transportClientInterceptors) { @@ -89,7 +84,6 @@ public synchronized void registerTransportClientInterceptor(ITransportClientInte this.transportClientInterceptors = new ArrayList<>(); } - logger.info("registerTransportClientInterceptor in Transport client"); this.transportClientInterceptors.add(transportClientInterceptor); } 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" } + } } ] }