Skip to content

Commit 4896510

Browse files
jeet1995jeet1995
authored
Default enablement of Threshold-Based Availability Strategy with Per-Partition Automatic Failover. (Azure#45267)
* Adding default enablement of hedging. * Wiring effective e2e policy cfg for diagnostics visibility. * Updated CHANGELOG.md * Clean up * Clean up * Fixing tests. * Fixing tests. * Fixing bug where `QueryPlan` call is indefinitely retried. * Fixing bug where `QueryPlan` call is indefinitely retried. * Fixing bug where `QueryPlan` call is indefinitely retried. * Fixing bug where `QueryPlan` call is indefinitely retried. * Fixing bug where `QueryPlan` call is indefinitely retried. * Avoid NPEs in PPAF flow. * Avoid NPEs in PPAF flow. * Addressing review comments. * Addressing review comments. * Addressing review comments. * Addressing review comments. * Addressing review comments.
1 parent 025b4e1 commit 4896510

File tree

8 files changed

+492
-119
lines changed

8 files changed

+492
-119
lines changed

‎sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/EndToEndTimeOutWithAvailabilityTest.java

Lines changed: 261 additions & 90 deletions
Large diffs are not rendered by default.

‎sdk/cosmos/azure-cosmos/CHANGELOG.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,10 +13,12 @@
1313
* Fixed diagnostics issue where operations in Gateway mode hitting end-to-end timeout would not capture diagnostics correctly. - [PR 44099](https://github.com/Azure/azure-sdk-for-java/pull/44099)
1414
* Enabled `excludeRegions` to be applied for `QueryPlan` calls. - [PR 45196](https://github.com/Azure/azure-sdk-for-java/pull/45196)
1515
* Fixed the behavior to correctly perform partition-level failover through circuit breaker for operations enabled with threshold-based availability strategy. - [PR 45244](https://github.com/Azure/azure-sdk-for-java/pull/45244)
16+
* Fixed an issue where `QueryPlan` calls are indefinitely retried in the same region when there are connectivity or response delays with / from the Gateway. - [PR 45267](https://github.com/Azure/azure-sdk-for-java/pull/45267)
1617

1718
#### Other Changes
1819
* Added the `vectorIndexShardKeys` to the vectorIndexSpec for QuantizedFlat and DiskANN vector search. - [PR 44007](https://github.com/Azure/azure-sdk-for-java/pull/44007)
1920
* Added user agent suffixing if Per-Partition Automatic Failover or Per-Partition Circuit Breaker are enabled at client scope. - [PR 45197](https://github.com/Azure/azure-sdk-for-java/pull/45197)
21+
* Enable threshold-based availability strategy for reads by default when Per-Partition Automatic Failover is also enabled. - [PR 45267](https://github.com/Azure/azure-sdk-for-java/pull/45267)
2022

2123
### 4.68.0 (2025-03-20)
2224

‎sdk/cosmos/azure-cosmos/docs/TimeoutAndRetriesConfig.md

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,3 +33,17 @@
3333
| 410 | 1000 | NO | N/A | N/A | N/A | 1 | N/A | |
3434
| 410 | 1002 | NO | N/A | N/A | N/A | 1 | N/A | Only applies to `Query`, `ChangeFeed` |
3535
| 400 | 1001 | NO | N/A | N/A | N/A | 1 | N/A | |
36+
37+
### Per-Partition Automatic Failover (PPAF) defaults
38+
39+
With PPAF enabled, the SDK will also enable threshold-based availability strategy for item-based non-write operations (readItem, readMany, readAll, queryItems etc.) with defaults as below:
40+
41+
#### Threshold-based availability strategy defaults
42+
43+
NOTE: 6s was chosen as in `Direct` Connection Mode, the connect timeout and network request timeout are 5s. This will allow the SDK to do at least 1 in-region retry. In `Gateway` connection mode, the Gateway performs the in-region retries on behalf of the SDK within the same time bound.
44+
45+
| Connection Mode | End-to-end timeout | Threshold duration | Threshold step duration |
46+
|-----------------|--------------------|--------------------|-------------------------|
47+
| Direct | 6s | 1s | 500ms |
48+
| Gateway | 6s | 1s | 500ms |
49+

‎sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/Configs.java

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -304,6 +304,14 @@ public class Configs {
304304
private static final String E2E_TIMEOUT_ERROR_HIT_TIME_WINDOW_IN_SECONDS_FOR_PPAF = "COSMOS.E2E_TIMEOUT_ERROR_HIT_TIME_WINDOW_IN_SECONDS_FOR_PPAF";
305305
private static final String E2E_TIMEOUT_ERROR_HIT_TIME_WINDOW_IN_SECONDS_FOR_PPAF_VARIABLE = "COSMOS_E2E_TIMEOUT_ERROR_HIT_TIME_WINDOW_IN_SECONDS_FOR_PPAF";
306306

307+
private static final String DEFAULT_IS_READ_AVAILABILITY_STRATEGY_ENABLED_WITH_PPAF = "true";
308+
private static final String IS_READ_AVAILABILITY_STRATEGY_ENABLED_WITH_PPAF = "COSMOS.IS_READ_AVAILABILITY_STRATEGY_ENABLED_WITH_PPAF";
309+
private static final String IS_READ_AVAILABILITY_STRATEGY_ENABLED_WITH_PPAF_VARIABLE = "COSMOS_IS_READ_AVAILABILITY_STRATEGY_ENABLED_WITH_PPAF";
310+
311+
private static final int DEFAULT_WARN_LEVEL_LOGGING_THRESHOLD_FOR_PPAF = 25;
312+
private static final String WARN_LEVEL_LOGGING_THRESHOLD_FOR_PPAF = "COSMOS.WARN_LEVEL_LOGGING_THRESHOLD_FOR_PPAF";
313+
private static final String WARN_LEVEL_LOGGING_THRESHOLD_FOR_PPAF_VARIABLE = "COSMOS_WARN_LEVEL_LOGGING_THRESHOLD_FOR_PPAF_VARIABLE";
314+
307315
private static final String COSMOS_DISABLE_IMDS_ACCESS = "COSMOS.DISABLE_IMDS_ACCESS";
308316
private static final String COSMOS_DISABLE_IMDS_ACCESS_VARIABLE = "COSMOS_DISABLE_IMDS_ACCESS";
309317
private static final boolean COSMOS_DISABLE_IMDS_ACCESS_DEFAULT = false;
@@ -1167,4 +1175,24 @@ public static String getEmulatorHost() {
11671175
emptyToNull(System.getenv().get(EMULATOR_HOST_VARIABLE)),
11681176
DEFAULT_EMULATOR_HOST));
11691177
}
1178+
1179+
public static boolean isReadAvailabilityStrategyEnabledWithPpaf() {
1180+
String isReadAvailabilityStrategyEnabledWithPpaf = System.getProperty(
1181+
IS_READ_AVAILABILITY_STRATEGY_ENABLED_WITH_PPAF,
1182+
firstNonNull(
1183+
emptyToNull(System.getenv().get(IS_READ_AVAILABILITY_STRATEGY_ENABLED_WITH_PPAF_VARIABLE)),
1184+
DEFAULT_IS_READ_AVAILABILITY_STRATEGY_ENABLED_WITH_PPAF));
1185+
1186+
return Boolean.parseBoolean(isReadAvailabilityStrategyEnabledWithPpaf);
1187+
}
1188+
1189+
public static int getWarnLevelLoggingThresholdForPpaf() {
1190+
String warnLevelLoggingThresholdForPpaf = System.getProperty(
1191+
WARN_LEVEL_LOGGING_THRESHOLD_FOR_PPAF,
1192+
firstNonNull(
1193+
emptyToNull(System.getenv().get(WARN_LEVEL_LOGGING_THRESHOLD_FOR_PPAF_VARIABLE)),
1194+
String.valueOf(DEFAULT_WARN_LEVEL_LOGGING_THRESHOLD_FOR_PPAF)));
1195+
1196+
return Integer.parseInt(warnLevelLoggingThresholdForPpaf);
1197+
}
11701198
}

‎sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java

Lines changed: 93 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import com.azure.cosmos.CosmosDiagnostics;
1515
import com.azure.cosmos.CosmosDiagnosticsContext;
1616
import com.azure.cosmos.CosmosEndToEndOperationLatencyPolicyConfig;
17+
import com.azure.cosmos.CosmosEndToEndOperationLatencyPolicyConfigBuilder;
1718
import com.azure.cosmos.CosmosException;
1819
import com.azure.cosmos.CosmosItemSerializer;
1920
import com.azure.cosmos.CosmosOperationPolicy;
@@ -192,6 +193,7 @@ public class RxDocumentClientImpl implements AsyncDocumentClient, IAuthorization
192193

193194
private static final String DUMMY_SQL_QUERY = "this is dummy and only used in creating " +
194195
"ParallelDocumentQueryExecutioncontext, but not used";
196+
195197
private final static ObjectMapper mapper = Utils.getSimpleObjectMapper();
196198
private final CosmosItemSerializer defaultCustomSerializer;
197199
private final static Logger logger = LoggerFactory.getLogger(RxDocumentClientImpl.class);
@@ -266,7 +268,8 @@ public class RxDocumentClientImpl implements AsyncDocumentClient, IAuthorization
266268
private final boolean sessionCapturingDisabled;
267269
private final boolean isRegionScopedSessionCapturingEnabledOnClientOrSystemConfig;
268270
private List<CosmosOperationPolicy> operationPolicies;
269-
private AtomicReference<CosmosAsyncClient> cachedCosmosAsyncClientSnapshot;
271+
private final AtomicReference<CosmosAsyncClient> cachedCosmosAsyncClientSnapshot;
272+
private final CosmosEndToEndOperationLatencyPolicyConfig ppafEnforcedE2ELatencyPolicyConfigForReads;
270273

271274
public RxDocumentClientImpl(URI serviceEndpoint,
272275
String masterKeyOrResourceToken,
@@ -600,6 +603,10 @@ private RxDocumentClientImpl(URI serviceEndpoint,
600603
this.queryPlanCache = new ConcurrentHashMap<>();
601604
this.apiType = apiType;
602605
this.clientTelemetryConfig = clientTelemetryConfig;
606+
this.ppafEnforcedE2ELatencyPolicyConfigForReads = evaluatePpafEnforcedE2eLatencyPolicyCfgForReads(
607+
this.globalPartitionEndpointManagerForPerPartitionAutomaticFailover,
608+
this.connectionPolicy
609+
);
603610
} catch (RuntimeException e) {
604611
logger.error("unexpected failure in initializing client.", e);
605612
close();
@@ -2466,7 +2473,7 @@ private Mono<ResourceResponse<Document>> createDocumentCore(
24662473
crossRegionAvailabilityContextForRxDocumentServiceRequest),
24672474
requestRetryPolicy),
24682475
scopedDiagnosticsFactory
2469-
), requestReference);
2476+
), requestReference, endToEndPolicyConfig);
24702477
}
24712478

24722479
private Mono<ResourceResponse<Document>> createDocumentInternal(
@@ -2576,7 +2583,10 @@ private static <T> Mono<T> getPointOperationResponseMonoWithE2ETimeout(
25762583

25772584
private <T> Mono<T> handleCircuitBreakingFeedbackForPointOperation(
25782585
Mono<T> response,
2579-
AtomicReference<RxDocumentServiceRequest> requestReference) {
2586+
AtomicReference<RxDocumentServiceRequest> requestReference,
2587+
CosmosEndToEndOperationLatencyPolicyConfig effectiveEndToEndPolicyConfig) {
2588+
2589+
applyEndToEndLatencyPolicyCfgToRequestContext(requestReference.get(), effectiveEndToEndPolicyConfig);
25802590

25812591
return response
25822592
.doOnSuccess(ignore -> {
@@ -2769,6 +2779,23 @@ private static CosmosException getNegativeTimeoutException(CosmosDiagnostics cos
27692779
return exception;
27702780
}
27712781

2782+
private static void applyEndToEndLatencyPolicyCfgToRequestContext(RxDocumentServiceRequest rxDocumentServiceRequest, CosmosEndToEndOperationLatencyPolicyConfig effectiveEndToEndPolicyConfig) {
2783+
2784+
if (rxDocumentServiceRequest == null) {
2785+
return;
2786+
}
2787+
2788+
if (rxDocumentServiceRequest.requestContext == null) {
2789+
return;
2790+
}
2791+
2792+
if (effectiveEndToEndPolicyConfig == null) {
2793+
return;
2794+
}
2795+
2796+
rxDocumentServiceRequest.requestContext.setEndToEndOperationLatencyPolicyConfig(effectiveEndToEndPolicyConfig);
2797+
}
2798+
27722799
@Override
27732800
public Mono<ResourceResponse<Document>> upsertDocument(String collectionLink, Object document,
27742801
RequestOptions options, boolean disableAutomaticIdGeneration) {
@@ -2831,7 +2858,7 @@ private Mono<ResourceResponse<Document>> upsertDocumentCore(
28312858
requestReference,
28322859
crossRegionAvailabilityContextForRequest),
28332860
finalRetryPolicyInstance),
2834-
scopedDiagnosticsFactory), requestReference);
2861+
scopedDiagnosticsFactory), requestReference, endToEndPolicyConfig);
28352862
}
28362863

28372864
private Mono<ResourceResponse<Document>> upsertDocumentInternal(
@@ -2974,7 +3001,7 @@ private Mono<ResourceResponse<Document>> replaceDocumentCore(
29743001
requestReference,
29753002
crossRegionAvailabilityContextForRequest),
29763003
requestRetryPolicy),
2977-
scopedDiagnosticsFactory), requestReference);
3004+
scopedDiagnosticsFactory), requestReference, endToEndPolicyConfig);
29783005
}
29793006

29803007
private Mono<ResourceResponse<Document>> replaceDocumentInternal(
@@ -3056,7 +3083,7 @@ private Mono<ResourceResponse<Document>> replaceDocumentCore(
30563083
clientContextOverride,
30573084
requestReference,
30583085
crossRegionAvailabilityContextForRequest),
3059-
requestRetryPolicy), requestReference);
3086+
requestRetryPolicy), requestReference, cosmosEndToEndOperationLatencyPolicyConfig);
30603087
}
30613088

30623089
private Mono<ResourceResponse<Document>> replaceDocumentInternal(
@@ -3232,7 +3259,17 @@ private CosmosEndToEndOperationLatencyPolicyConfig getEffectiveEndToEndOperation
32323259
return null;
32333260
}
32343261

3235-
return this.cosmosEndToEndOperationLatencyPolicyConfig;
3262+
if (this.cosmosEndToEndOperationLatencyPolicyConfig != null) {
3263+
return this.cosmosEndToEndOperationLatencyPolicyConfig;
3264+
}
3265+
3266+
// If request options level and client-level e2e latency policy config,
3267+
// rely on PPAF enforced defaults
3268+
if (operationType.isReadOnlyOperation()) {
3269+
return this.ppafEnforcedE2ELatencyPolicyConfigForReads;
3270+
}
3271+
3272+
return null;
32363273
}
32373274

32383275
@Override
@@ -3295,7 +3332,7 @@ private Mono<ResourceResponse<Document>> patchDocumentCore(
32953332
requestReference,
32963333
crossRegionAvailabilityContextForRequest),
32973334
documentClientRetryPolicy),
3298-
scopedDiagnosticsFactory), requestReference);
3335+
scopedDiagnosticsFactory), requestReference, cosmosEndToEndOperationLatencyPolicyConfig);
32993336
}
33003337

33013338
private Mono<ResourceResponse<Document>> patchDocumentInternal(
@@ -3504,7 +3541,7 @@ private Mono<ResourceResponse<Document>> deleteDocumentCore(
35043541
requestReference,
35053542
crossRegionAvailabilityContextForRequest),
35063543
requestRetryPolicy),
3507-
scopedDiagnosticsFactory), requestReference);
3544+
scopedDiagnosticsFactory), requestReference, endToEndPolicyConfig);
35083545
}
35093546

35103547
private Mono<ResourceResponse<Document>> deleteDocumentInternal(
@@ -3692,7 +3729,7 @@ private Mono<ResourceResponse<Document>> readDocumentCore(
36923729
crossRegionAvailabilityContextForRequest),
36933730
retryPolicyInstance),
36943731
scopedDiagnosticsFactory
3695-
), requestReference);
3732+
), requestReference, endToEndPolicyConfig);
36963733
}
36973734

36983735
private Mono<ResourceResponse<Document>> readDocumentInternal(
@@ -4883,7 +4920,7 @@ public Mono<CosmosBatchResponse> executeBatchRequest(String collectionLink,
48834920
requestReference), documentClientRetryPolicy),
48844921
scopedDiagnosticsFactory
48854922
),
4886-
requestReference);
4923+
requestReference, endToEndPolicyConfig);
48874924
}
48884925

48894926
private Mono<StoredProcedureResponse> executeStoredProcedureInternal(String storedProcedureLink,
@@ -7145,6 +7182,49 @@ private static boolean isNonTransientResultForHedging(int statusCode, int subSta
71457182
return false;
71467183
}
71477184

7185+
private static CosmosEndToEndOperationLatencyPolicyConfig evaluatePpafEnforcedE2eLatencyPolicyCfgForReads(
7186+
GlobalPartitionEndpointManagerForPerPartitionAutomaticFailover globalPartitionEndpointManagerForPerPartitionAutomaticFailover,
7187+
ConnectionPolicy connectionPolicy) {
7188+
7189+
if (!globalPartitionEndpointManagerForPerPartitionAutomaticFailover.isPerPartitionAutomaticFailoverEnabled()) {
7190+
return null;
7191+
}
7192+
7193+
if (Configs.isReadAvailabilityStrategyEnabledWithPpaf()) {
7194+
7195+
if (connectionPolicy.getConnectionMode() == ConnectionMode.DIRECT) {
7196+
Duration networkRequestTimeout = connectionPolicy.getTcpNetworkRequestTimeout();
7197+
7198+
checkNotNull(networkRequestTimeout, "Argument 'networkRequestTimeout' cannot be null!");
7199+
7200+
Duration overallE2eLatencyTimeout = networkRequestTimeout.plus(Utils.ONE_SECOND);
7201+
Duration threshold = Utils.min(networkRequestTimeout.dividedBy(2), Utils.ONE_SECOND);
7202+
Duration thresholdStep = Utils.min(threshold.dividedBy(2), Utils.HALF_SECOND);
7203+
7204+
return new CosmosEndToEndOperationLatencyPolicyConfigBuilder(overallE2eLatencyTimeout)
7205+
.availabilityStrategy(new ThresholdBasedAvailabilityStrategy(threshold, thresholdStep))
7206+
.build();
7207+
} else {
7208+
7209+
Duration httpNetworkRequestTimeout = connectionPolicy.getHttpNetworkRequestTimeout();
7210+
7211+
checkNotNull(httpNetworkRequestTimeout, "Argument 'httpNetworkRequestTimeout' cannot be null!");
7212+
7213+
// 6s was chosen to accommodate for control-plane hot path read timeout retries (like QueryPlan / PartitionKeyRange)
7214+
Duration overallE2eLatencyTimeout = Utils.min(Utils.SIX_SECONDS, httpNetworkRequestTimeout);
7215+
7216+
Duration threshold = Utils.min(overallE2eLatencyTimeout.dividedBy(2), Utils.ONE_SECOND);
7217+
Duration thresholdStep = Utils.min(threshold.dividedBy(2), Utils.HALF_SECOND);
7218+
7219+
return new CosmosEndToEndOperationLatencyPolicyConfigBuilder(overallE2eLatencyTimeout)
7220+
.availabilityStrategy(new ThresholdBasedAvailabilityStrategy(threshold, thresholdStep))
7221+
.build();
7222+
}
7223+
}
7224+
7225+
return null;
7226+
}
7227+
71487228
private DiagnosticsClientContext getEffectiveClientContext(DiagnosticsClientContext clientContextOverride) {
71497229
if (clientContextOverride != null) {
71507230
return clientContextOverride;
@@ -7261,6 +7341,8 @@ private <T> Mono<T> executeFeedOperationWithAvailabilityStrategy(
72617341
this.getEffectiveEndToEndOperationLatencyPolicyConfig(
72627342
req.requestContext.getEndToEndOperationLatencyPolicyConfig(), resourceType, operationType);
72637343

7344+
req.requestContext.setEndToEndOperationLatencyPolicyConfig(endToEndPolicyConfig);
7345+
72647346
List<String> initialExcludedRegions = req.requestContext.getExcludeRegions();
72657347
List<String> orderedApplicableRegionsForSpeculation = this.getApplicableRegionsForSpeculation(
72667348
endToEndPolicyConfig,

‎sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/Utils.java

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,10 @@ public class Utils {
7878
public static final Base64.Decoder Base64Decoder = Base64.getDecoder();
7979
public static final Base64.Encoder Base64UrlEncoder = Base64.getUrlEncoder();
8080

81+
public static final Duration ONE_SECOND = Duration.ofSeconds(1);
82+
public static final Duration HALF_SECOND = Duration.ofMillis(500);
83+
public static final Duration SIX_SECONDS = Duration.ofSeconds(6);
84+
8185
private static final ObjectMapper simpleObjectMapperAllowingDuplicatedProperties =
8286
createAndInitializeObjectMapper(true);
8387
private static final ObjectMapper simpleObjectMapperDisallowingDuplicatedProperties =
@@ -808,4 +812,14 @@ public static boolean shouldAllowUnquotedControlChars() {
808812

809813
return Boolean.parseBoolean(shouldAllowUnquotedControlCharsConfig);
810814
}
815+
816+
public static Duration min(Duration duration1, Duration duration2) {
817+
if (duration1 == null) {
818+
return duration2;
819+
} else if (duration2 == null) {
820+
return duration1;
821+
} else {
822+
return duration1.compareTo(duration2) < 0 ? duration1 : duration2;
823+
}
824+
}
811825
}

0 commit comments

Comments
 (0)