Skip to content

Commit dca0ce5

Browse files
FabianMeiswinkelFabianMeiswinkel
authored
Ensuring effective ReadConsistencyStrategy is returned from DiagnosticsCtx (#45645)
* Ensuring effective ReadConsistencyStrategy is returned from DiagnosticsCtx * Update CosmosLatestCommittedItemTests.java * Update CHANGELOG.md * Update CHANGELOG.md * Fixing test regressions
1 parent 182bdae commit dca0ce5

7 files changed

Lines changed: 98 additions & 36 deletions

File tree

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

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -137,7 +137,9 @@ public void queryItemsWithLatestCommitted() throws JsonProcessingException {
137137

138138
String id = UUID.randomUUID().toString();
139139
ObjectNode properties = getDocumentDefinition(id, id);
140-
container.createItem(properties);
140+
CosmosItemResponse<ObjectNode> createResponse = container.createItem(properties);
141+
assertThat(createResponse.getDiagnostics().getDiagnosticsContext().getEffectiveReadConsistencyStrategy())
142+
.isEqualTo(ReadConsistencyStrategy.DEFAULT);
141143

142144
String query = String.format("SELECT * from c where c.id = '%s'", id);
143145
CosmosQueryRequestOptions requestOptions = new CosmosQueryRequestOptions()
@@ -211,7 +213,9 @@ public void readItemWithLatestCommitted() throws JsonProcessingException {
211213

212214
String id = UUID.randomUUID().toString();
213215
ObjectNode properties = getDocumentDefinition(id, id);
214-
container.createItem(properties);
216+
CosmosItemResponse<ObjectNode> createResponse = container.createItem(properties);
217+
assertThat(createResponse.getDiagnostics().getDiagnosticsContext().getEffectiveReadConsistencyStrategy())
218+
.isEqualTo(ReadConsistencyStrategy.DEFAULT);
215219

216220
CosmosItemRequestOptions requestOptions = new CosmosItemRequestOptions()
217221
.setReadConsistencyStrategy(ReadConsistencyStrategy.LATEST_COMMITTED);
@@ -260,11 +264,15 @@ public void readManyWithLatestCommitted() throws JsonProcessingException {
260264
String pk = UUID.randomUUID().toString();
261265
String id1 = UUID.randomUUID().toString();
262266
ObjectNode properties1 = getDocumentDefinition(id1, pk);
263-
container.createItem(properties1);
267+
CosmosItemResponse<ObjectNode> createResponse = container.createItem(properties1);
268+
assertThat(createResponse.getDiagnostics().getDiagnosticsContext().getEffectiveReadConsistencyStrategy())
269+
.isEqualTo(ReadConsistencyStrategy.DEFAULT);
264270

265271
String id2 = UUID.randomUUID().toString();
266272
ObjectNode properties2 = getDocumentDefinition(id2, pk);
267-
container.createItem(properties2);
273+
createResponse = container.createItem(properties2);
274+
assertThat(createResponse.getDiagnostics().getDiagnosticsContext().getEffectiveReadConsistencyStrategy())
275+
.isEqualTo(ReadConsistencyStrategy.DEFAULT);
268276

269277
List<CosmosItemIdentity> identities = new ArrayList<>();
270278
identities.add(new CosmosItemIdentity(new PartitionKey(pk), id1));
@@ -317,7 +325,9 @@ public void readManyWithLatestCommitted_SingleItem() throws JsonProcessingExcept
317325
String pk = UUID.randomUUID().toString();
318326
String id1 = UUID.randomUUID().toString();
319327
ObjectNode properties1 = getDocumentDefinition(id1, pk);
320-
container.createItem(properties1);
328+
CosmosItemResponse<ObjectNode> createResponse = container.createItem(properties1);
329+
assertThat(createResponse.getDiagnostics().getDiagnosticsContext().getEffectiveReadConsistencyStrategy())
330+
.isEqualTo(ReadConsistencyStrategy.DEFAULT);
321331

322332
List<CosmosItemIdentity> identities = new ArrayList<>();
323333
identities.add(new CosmosItemIdentity(new PartitionKey(pk), id1));
@@ -369,7 +379,9 @@ public void queryChangeFeedWithLatestCommitted() throws JsonProcessingException
369379
String pk = UUID.randomUUID().toString();
370380
String id1 = UUID.randomUUID().toString();
371381
ObjectNode properties1 = getDocumentDefinition(id1, pk);
372-
container.createItem(properties1);
382+
CosmosItemResponse<ObjectNode> createResponse = container.createItem(properties1);
383+
assertThat(createResponse.getDiagnostics().getDiagnosticsContext().getEffectiveReadConsistencyStrategy())
384+
.isEqualTo(ReadConsistencyStrategy.DEFAULT);
373385

374386
List<CosmosItemIdentity> identities = new ArrayList<>();
375387
identities.add(new CosmosItemIdentity(new PartitionKey(pk), id1));

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,14 +7,14 @@
77
#### Breaking Changes
88

99
#### Bugs Fixed
10+
* Fixed an issue returning incorrect `ReadConsistencyStrategy` in `CosmosDiagnosticsContext.getEffectiveReadConsistencyStrategy()` for change feed operations in some cases - [PR 45645](https://github.com/Azure/azure-sdk-for-java/pull/45645)
1011

1112
#### Other Changes
1213

1314
### 4.71.0 (2025-06-04)
1415

1516
#### Features Added
1617
* Added `GetEffectiveReadConsistencyLevel` method on the `CosmosDiagnosticsContext`. - [PR 45414](https://github.com/Azure/azure-sdk-for-java/pull/45414)
17-
* Added Weighted RRF for Hybrid and Full Text Search queries - [PR 45328](https://github.com/Azure/azure-sdk-for-java/pull/45328)
1818

1919
#### Bugs Fixed
2020
* Fixed an issue ignoring client-level diagnostic thresholds for point read operations in some cases - [PR 45488](https://github.com/Azure/azure-sdk-for-java/pull/45488)
@@ -46,7 +46,7 @@
4646
### 4.69.0 (2025-05-14)
4747

4848
#### Features Added
49-
* Added Beta public API to enable http2. - See [PR 43123](https://github.com/Azure/azure-sdk-for-java/pull/43123)
49+
* Added Beta public API to enable http2. - See [PR 45294](https://github.com/Azure/azure-sdk-for-java/pull/45294)
5050
* Added API to allow customers to wrap/extend `CosmosAsyncContainer` - [PR 43724](https://github.com/Azure/azure-sdk-for-java/pull/43724) and [PR 45087](https://github.com/Azure/azure-sdk-for-java/pull/45087)
5151
* Added Per-Partition Automatic Failover which enables failover for writes at per-partition level for Single-Write Multi-Region accounts. - [PR 44099](https://github.com/Azure/azure-sdk-for-java/pull/44099)
5252
* Added Beta public API to allow defining the consistency behavior for read / query / change feed operations independent of the chosen account-level consistency level. **NOTE: This API is still in preview mode and can only be used when using DIRECT connection mode.** - See [PR 45161](https://github.com/Azure/azure-sdk-for-java/pull/45161)

‎sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosAsyncClient.java‎

Lines changed: 11 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,11 @@ public final class CosmosAsyncClient implements Closeable {
8989
.CosmosClientTelemetryConfigHelper
9090
.getCosmosClientTelemetryConfigAccessor();
9191

92+
private static final ImplementationBridgeHelpers.ReadConsistencyStrategyHelper.ReadConsistencyStrategyAccessor
93+
readConsistencyStrategyAccessor = ImplementationBridgeHelpers
94+
.ReadConsistencyStrategyHelper
95+
.getReadConsistencyStrategyAccessor();
96+
9297
private final static Function<CosmosAsyncContainer, CosmosAsyncContainer> DEFAULT_CONTAINER_FACTORY =
9398
(originalContainer) -> originalContainer;
9499

@@ -748,23 +753,12 @@ ReadConsistencyStrategy getEffectiveReadConsistencyStrategy(
748753
OperationType operationType,
749754
ReadConsistencyStrategy desiredReadConsistencyStrategyOfOperation) {
750755

751-
if (resourceType != ResourceType.Document) {
752-
return ReadConsistencyStrategy.DEFAULT;
753-
}
754-
755-
if (operationType.isWriteOperation()) {
756-
return ReadConsistencyStrategy.DEFAULT;
757-
}
758-
759-
if (desiredReadConsistencyStrategyOfOperation != null) {
760-
return desiredReadConsistencyStrategyOfOperation;
761-
}
762-
763-
if (this.readConsistencyStrategy != null) {
764-
return readConsistencyStrategy;
765-
}
766-
767-
return ReadConsistencyStrategy.DEFAULT;
756+
return readConsistencyStrategyAccessor.getEffectiveReadConsistencyStrategy(
757+
resourceType,
758+
operationType,
759+
desiredReadConsistencyStrategyOfOperation,
760+
this.readConsistencyStrategy
761+
);
768762
}
769763

770764
CosmosDiagnosticsThresholds getEffectiveDiagnosticsThresholds(

‎sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosDiagnosticsContext.java‎

Lines changed: 12 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ public final class CosmosDiagnosticsContext {
5959
private final OperationType operationType;
6060
private final String operationTypeString;
6161
private final ConsistencyLevel consistencyLevel;
62-
private final ReadConsistencyStrategy readConsistencyStrategy;
62+
private ReadConsistencyStrategy readConsistencyStrategy;
6363
private final ConcurrentLinkedDeque<CosmosDiagnostics> diagnostics;
6464
private final Integer maxItemCount;
6565
private final CosmosDiagnosticsThresholds thresholds;
@@ -91,7 +91,7 @@ public final class CosmosDiagnosticsContext {
9191

9292
private final Integer sequenceNumber;
9393

94-
private String queryStatement;
94+
private final String queryStatement;
9595

9696
private Long opCountPerEvaluation;
9797
private Long opRetriedCountPerEvaluation;
@@ -257,10 +257,6 @@ public ConsistencyLevel getEffectiveConsistencyLevel() {
257257
*/
258258
@Beta(value = Beta.SinceVersion.V4_71_0, warningText = Beta.PREVIEW_SUBJECT_TO_CHANGE_WARNING)
259259
public ReadConsistencyStrategy getEffectiveReadConsistencyStrategy() {
260-
if (this.requestOptions != null && this.requestOptions.getReadConsistencyStrategy() != null) {
261-
return this.requestOptions.getReadConsistencyStrategy();
262-
}
263-
264260
return this.readConsistencyStrategy;
265261
}
266262

@@ -981,8 +977,12 @@ OverridableRequestOptions getRequestOptions() {
981977
return this.requestOptions;
982978
}
983979

984-
void setRequestOptions(OverridableRequestOptions requestOptions) {
980+
void setRequestOptions(
981+
OverridableRequestOptions requestOptions,
982+
ReadConsistencyStrategy newEffectiveReadConsistencyStrategy) {
983+
985984
this.requestOptions = requestOptions;
985+
this.readConsistencyStrategy = newEffectiveReadConsistencyStrategy;
986986
}
987987

988988
///////////////////////////////////////////////////////////////////////////////////////////
@@ -1039,9 +1039,12 @@ public OverridableRequestOptions getRequestOptions(CosmosDiagnosticsContext ctx)
10391039
}
10401040

10411041
@Override
1042-
public void setRequestOptions(CosmosDiagnosticsContext ctx, OverridableRequestOptions requestOptions) {
1042+
public void setRequestOptions(
1043+
CosmosDiagnosticsContext ctx,
1044+
OverridableRequestOptions requestOptions,
1045+
ReadConsistencyStrategy newEffectiveReadConsistencyStrategy) {
10431046
checkNotNull(ctx, "Argument 'ctx' must not be null.");
1044-
ctx.setRequestOptions(requestOptions);
1047+
ctx.setRequestOptions(requestOptions, newEffectiveReadConsistencyStrategy);
10451048
}
10461049

10471050
@Override

‎sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/ReadConsistencyStrategy.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,8 @@
44
package com.azure.cosmos;
55

66
import com.azure.cosmos.implementation.ImplementationBridgeHelpers;
7+
import com.azure.cosmos.implementation.OperationType;
8+
import com.azure.cosmos.implementation.ResourceType;
79
import com.azure.cosmos.util.Beta;
810
import com.fasterxml.jackson.annotation.JsonValue;
911

@@ -100,6 +102,31 @@ static void initialize() {
100102
public ReadConsistencyStrategy createFromServiceSerializedFormat(String serviceSerializedFormat) {
101103
return ReadConsistencyStrategy.fromServiceSerializedFormat(serviceSerializedFormat);
102104
}
105+
106+
public ReadConsistencyStrategy getEffectiveReadConsistencyStrategy(
107+
ResourceType resourceType,
108+
OperationType operationType,
109+
ReadConsistencyStrategy desiredReadConsistencyStrategyOfOperation,
110+
ReadConsistencyStrategy clientLevelReadConsistencyStrategy) {
111+
112+
if (resourceType != ResourceType.Document) {
113+
return ReadConsistencyStrategy.DEFAULT;
114+
}
115+
116+
if (operationType.isWriteOperation()) {
117+
return ReadConsistencyStrategy.DEFAULT;
118+
}
119+
120+
if (desiredReadConsistencyStrategyOfOperation != null) {
121+
return desiredReadConsistencyStrategyOfOperation;
122+
}
123+
124+
if (clientLevelReadConsistencyStrategy != null) {
125+
return clientLevelReadConsistencyStrategy;
126+
}
127+
128+
return ReadConsistencyStrategy.DEFAULT;
129+
}
103130
}
104131
);
105132
}

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

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -905,7 +905,9 @@ CosmosDiagnosticsContext create(
905905

906906
OverridableRequestOptions getRequestOptions(CosmosDiagnosticsContext ctx);
907907

908-
void setRequestOptions(CosmosDiagnosticsContext ctx, OverridableRequestOptions requestOptions);
908+
void setRequestOptions(CosmosDiagnosticsContext ctx,
909+
OverridableRequestOptions requestOptions,
910+
ReadConsistencyStrategy updatedEffectiveReadConsistencyStrategy);
909911

910912
CosmosDiagnosticsSystemUsageSnapshot createSystemUsageSnapshot(
911913
String cpu,
@@ -1892,6 +1894,11 @@ public static ReadConsistencyStrategyAccessor getReadConsistencyStrategyAccessor
18921894

18931895
public interface ReadConsistencyStrategyAccessor {
18941896
ReadConsistencyStrategy createFromServiceSerializedFormat(String serviceSerializedFormat);
1897+
ReadConsistencyStrategy getEffectiveReadConsistencyStrategy(
1898+
ResourceType resourceType,
1899+
OperationType operationType,
1900+
ReadConsistencyStrategy desiredReadConsistencyStrategyOfOperation,
1901+
ReadConsistencyStrategy clientLevelReadConsistencyStrategy);
18951902
}
18961903
}
18971904
}

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

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -183,6 +183,9 @@ public class RxDocumentClientImpl implements AsyncDocumentClient, IAuthorization
183183
private static final ImplementationBridgeHelpers.CosmosOperationDetailsHelper.CosmosOperationDetailsAccessor operationDetailsAccessor =
184184
ImplementationBridgeHelpers.CosmosOperationDetailsHelper.getCosmosOperationDetailsAccessor();
185185

186+
private static final ImplementationBridgeHelpers.ReadConsistencyStrategyHelper.ReadConsistencyStrategyAccessor readConsistencyStrategyAccessor =
187+
ImplementationBridgeHelpers.ReadConsistencyStrategyHelper.getReadConsistencyStrategyAccessor();
188+
186189
private static final String tempMachineId = "uuid:" + UUIDs.nonBlockingRandomUUID();
187190
private static final AtomicInteger activeClientsCnt = new AtomicInteger(0);
188191
private static final Map<String, Integer> clientMap = new ConcurrentHashMap<>();
@@ -4520,7 +4523,23 @@ public <T> Flux<FeedResponse<T>> queryDocumentChangeFeedFromPagedFlux(DocumentCo
45204523
throw(exception);
45214524
}
45224525
});
4523-
ctxAccessor.setRequestOptions(state.getDiagnosticsContextSnapshot(), optionsImpl);
4526+
4527+
ReadConsistencyStrategy requestLevelReadConsistencyStrategy = optionsImpl != null
4528+
? optionsImpl.getReadConsistencyStrategy()
4529+
: null;
4530+
4531+
ReadConsistencyStrategy effectiveReadConsistencyStrategy = readConsistencyStrategyAccessor
4532+
.getEffectiveReadConsistencyStrategy(
4533+
ResourceType.Document,
4534+
OperationType.ReadFeed,
4535+
requestLevelReadConsistencyStrategy,
4536+
this.readConsistencyStrategy);
4537+
4538+
ctxAccessor.setRequestOptions(
4539+
state.getDiagnosticsContextSnapshot(),
4540+
optionsImpl,
4541+
effectiveReadConsistencyStrategy);
4542+
45244543
return queryDocumentChangeFeed(collection, clonedOptions, classOfT);
45254544
}
45264545

0 commit comments

Comments
 (0)