From 248144ddd38f89ff64e71fff8a5a5d4329613de7 Mon Sep 17 00:00:00 2001 From: tvaron3 Date: Mon, 22 Apr 2024 15:41:15 -0700 Subject: [PATCH 01/16] quorumAckedLSN and currentReplicaSetSize added to diagnostics. replicaStatusList includes all replicas --- .../azure/cosmos/CosmosDiagnosticsTest.java | 18 +++++--- sdk/cosmos/azure-cosmos/CHANGELOG.md | 1 + .../directconnectivity/AddressSelector.java | 22 +++++++++- .../directconnectivity/ConsistencyWriter.java | 16 +++++-- .../directconnectivity/StoreReader.java | 26 +++++++---- .../StoreResultDiagnostics.java | 2 + .../directconnectivity/Uri.java | 44 +++++++++++++++++-- 7 files changed, 109 insertions(+), 20 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java index 22aeb65008b5..513286edbf1c 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java @@ -980,6 +980,11 @@ public void directDiagnosticsOnException() throws Exception { JsonNode replicaStatusList = storeResult.get("replicaStatusList"); assertThat(replicaStatusList.isArray()).isTrue(); assertThat(replicaStatusList.size()).isGreaterThan(0); + int quorumAcked = storeResult.get("quorumAckedLSN").asInt(-1); + assertThat(quorumAcked).isGreaterThan(0); + int currentReplicaSetSize = storeResult.get("currentReplicaSetSize").asInt(-1); + assertThat(currentReplicaSetSize).isGreaterThan(0); + } } @@ -1222,11 +1227,14 @@ private void validateRntbdStatistics(CosmosDiagnostics cosmosDiagnostics, assertThat(responseStatisticsList.size()).isGreaterThan(0); JsonNode storeResult = responseStatisticsList.get(0).get("storeResult"); assertThat(storeResult).isNotNull(); - + int replicaSetSize = storeResult.get("currentReplicaSetSize").asInt(-1); + assertThat(replicaSetSize).isGreaterThan(0); JsonNode replicaStatusList = storeResult.get("replicaStatusList"); assertThat(replicaStatusList.isArray()).isTrue(); - assertThat(replicaStatusList.size()).isGreaterThan(0); - + assertThat(replicaStatusList.size()).isEqualTo(replicaSetSize); + String replicaStatusTxt = replicaStatusList.get(0).asText(); + assertThat(replicaStatusTxt.contains("Primary")).isTrue(); + assertThat(replicaStatusTxt.contains("Connected")).isTrue(); // validate serviceEndpointStatistics JsonNode serviceEndpointStatistics = storeResult.get("serviceEndpointStatistics"); assertThat(serviceEndpointStatistics).isNotNull(); @@ -1502,14 +1510,14 @@ public void negativeE2ETimeoutWithQueryOperation() { TestItem testItem = TestItem.createNewItem(); container.createItem(testItem).block(); - + CosmosQueryRequestOptions requestOptions = new CosmosQueryRequestOptions(); requestOptions.setCosmosEndToEndOperationLatencyPolicyConfig( new CosmosEndToEndOperationLatencyPolicyConfigBuilder(Duration.ofSeconds(-1)).build() ); CosmosPagedFlux flux = container.readAllItems(requestOptions, ObjectNode.class); List results = flux.collectList().block(); - + fail("This should have failed with an exception"); } catch(OperationCancelledException cancelledException) { assertThat(cancelledException).isNotNull(); diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index ddcf1574edc0..badfca11091d 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -3,6 +3,7 @@ ### 4.59.0-beta.1 (Unreleased) #### Features Added +* Added diagnostic fields for `quorumAckedLSN` and `currentReplicaSetSize`. Changed `replicaStatusList` to include all replicas and more information. - See [PR 39807]() #### Breaking Changes diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java index b0b0810f0d75..4cd9e9c49306 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java @@ -14,6 +14,7 @@ import java.util.Arrays; import java.util.List; +import java.util.Map; import java.util.stream.Collectors; public class AddressSelector { @@ -31,7 +32,12 @@ public Mono> resolveAllUriAsync( boolean forceRefresh) { Mono> allReplicaAddressesObs = this.resolveAddressesAsync(request, forceRefresh); return allReplicaAddressesObs.map(allReplicaAddresses -> allReplicaAddresses.stream().filter(a -> includePrimary || !a.isPrimary()) - .map(a -> a.getPhysicalUri()).collect(Collectors.toList())); + .map(a -> { + if (includePrimary && a.isPrimary()) { + a.getPhysicalUri().setHealthStatusTuplePrimary(true); + } + return a.getPhysicalUri(); + }).collect(Collectors.toList())); } public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolean forceAddressRefresh) { @@ -45,6 +51,20 @@ public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolea }); } + public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolean forceAddressRefresh, Map replicaStatuses) { + Mono> replicaAddressesObs = this.resolveAddressesAsync(request, forceAddressRefresh); + return replicaAddressesObs.flatMap(replicaAddresses -> { + try { + replicaAddresses.forEach(replica -> { + replicaStatuses.put(replica.getPhysicalUri(), replica.getPhysicalUri().getHealthStatusDiagnosticString()); + }); + return Mono.just(AddressSelector.getPrimaryUri(request, replicaAddresses)); + } catch (Exception e) { + return Mono.error(e); + } + }); + } + public static Uri getPrimaryUri(RxDocumentServiceRequest request, List replicaAddresses) throws GoneException { AddressInformation primaryAddress = null; diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java index 1f9e1631c088..bea8b90027c1 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java @@ -37,7 +37,10 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; +import java.util.LinkedList; import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @@ -167,11 +170,16 @@ Mono writePrivateAsync( Mono> replicaAddressesObs = this.addressSelector.resolveAddressesAsync(request, forceRefresh); AtomicReference primaryURI = new AtomicReference<>(); AtomicReference> replicaStatusList = new AtomicReference<>(); + Map replicaStatuses = new ConcurrentHashMap<>(); return replicaAddressesObs.flatMap(replicaAddresses -> { try { List contactedReplicas = new ArrayList<>(); - replicaAddresses.forEach(replicaAddress -> contactedReplicas.add(replicaAddress.getPhysicalUri().getURI())); + replicaAddresses.forEach(replicaAddress -> { + Uri uri = replicaAddress.getPhysicalUri(); + contactedReplicas.add(uri.getURI()); + replicaStatuses.put(uri, uri.getHealthStatusDiagnosticString()); + }); BridgeInternal.setContactedReplicas(request.requestContext.cosmosDiagnostics, contactedReplicas); return Mono.just(AddressSelector.getPrimaryUri(request, replicaAddresses)); } catch (GoneException e) { @@ -196,8 +204,10 @@ Mono writePrivateAsync( } catch (Exception e) { return Mono.error(e); } - - replicaStatusList.set(Arrays.asList(primaryUri.getHealthStatusDiagnosticString())); + primaryUri.setHealthStatusTuplePrimary(true); + primaryUri.setHealthStatusTupleAttempting(true); + replicaStatuses.replace(primaryUri, primaryUri.getHealthStatusDiagnosticString()); + replicaStatusList.set(new ArrayList<>(replicaStatuses.values())); return this.transportClient.invokeResourceOperationAsync(primaryUri, request) .doOnError( diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java index 8e8c281fa0fd..ca420ec6c686 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java @@ -37,7 +37,10 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; @@ -216,16 +219,20 @@ private Flux> readFromReplicas(List resultCollect return Flux.error(new GoneException()); } List, Uri>> readStoreTasks = new ArrayList<>(); - + Map replicaStatuses = new HashMap<>(); + resolveApiResults.forEach(uri -> { + replicaStatuses.put(uri, uri.getHealthStatusDiagnosticString()); + }); List addressRandomPermutation = AddressEnumerator.getTransportAddresses(entity, resolveApiResults); // The health status of the Uri will change as the time goes by // what we really want to track is the health status snapshot at this moment - List replicaStatusList = - addressRandomPermutation - .stream() - .map(uri -> uri.getHealthStatusDiagnosticString()) - .collect(Collectors.toList()); + addressRandomPermutation.forEach(uri -> { + uri.setHealthStatusTupleAttempting(true); + replicaStatuses.replace(uri, uri.getHealthStatusDiagnosticString()); + }); + // how to add primary + List replicaStatusList = new ArrayList<>(replicaStatuses.values()); int startIndex = 0; @@ -563,9 +570,10 @@ private Mono readPrimaryInternalAsync( return Mono.error(new GoneException()); } + Map replicaStatuses = new ConcurrentHashMap<>(); Mono primaryUriObs = this.addressSelector.resolvePrimaryUriAsync( entity, - entity.requestContext.forceRefreshAddressCache); + entity.requestContext.forceRefreshAddressCache, replicaStatuses); AtomicReference> replicaStatusList = new AtomicReference<>(); @@ -588,7 +596,9 @@ private Mono readPrimaryInternalAsync( this.readFromStoreAsync( primaryUri, entity); - replicaStatusList.set(Arrays.asList(primaryUri.getHealthStatusDiagnosticString())); + primaryUri.setHealthStatusTuplePrimary(true); + replicaStatuses.replace(primaryUri, primaryUri.getHealthStatusDiagnosticString()); + replicaStatusList.set(new ArrayList<>(replicaStatuses.values())); return storeResponseObsAndUri.getLeft().flatMap( storeResponse -> { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResultDiagnostics.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResultDiagnostics.java index 6e84c7a3ab59..6770ea63d09a 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResultDiagnostics.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResultDiagnostics.java @@ -201,6 +201,8 @@ public void serialize(StoreResultDiagnostics storeResultDiagnostics, "storePhysicalAddress", storeResultDiagnostics.storePhysicalAddress == null ? null : storeResultDiagnostics.storePhysicalAddress.getURIAsString()); jsonGenerator.writeNumberField("lsn", storeResultDiagnostics.lsn); + jsonGenerator.writeNumberField("quorumAckedLSN",storeResultDiagnostics.quorumAckedLSN); + jsonGenerator.writeNumberField("currentReplicaSetSize", storeResultDiagnostics.currentReplicaSetSize); jsonGenerator.writeNumberField("globalCommittedLsn", storeResultDiagnostics.globalCommittedLSN); jsonGenerator.writeStringField("partitionKeyRangeId", storeResponseDiagnostics.getPartitionKeyRangeId()); jsonGenerator.writeBooleanField("isValid", storeResultDiagnostics.isValid); diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java index ca0ff0fefd8f..d4d3cd39033b 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java @@ -3,6 +3,7 @@ package com.azure.cosmos.implementation.directconnectivity; +import com.azure.cosmos.implementation.cpu.CpuMemoryReader; import com.azure.cosmos.implementation.directconnectivity.addressEnumerator.AddressEnumerator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -183,7 +184,25 @@ public boolean shouldRefreshHealthStatus() { } public String getHealthStatusDiagnosticString() { - return this.healthStatusTuple.get().diagnsoticString; + return this.healthStatusTuple.get().getDiagnsoticString(); + } + + public void setHealthStatusTupleAttempting(boolean attempting) { + this.healthStatusTuple.updateAndGet(previousStatusTuple -> + { + previousStatusTuple.setAttempting(attempting); + return previousStatusTuple; + } + ); + } + + public void setHealthStatusTuplePrimary(boolean primary) { + this.healthStatusTuple.updateAndGet(previousStatusTuple -> + { + previousStatusTuple.setPrimary(primary); + return previousStatusTuple; + } + ); } @Override @@ -230,12 +249,31 @@ public int getPriority() { } static class HealthStatusAndDiagnosticStringTuple { - private final String diagnsoticString; private final HealthStatus status; + private final String ATTEMPTING = "Attempting"; + private final String NOT_ATTEMPTING = "NotAttempting"; + private final String PRIMARY = "Primary"; + private final String SECONDARY = "Secondary"; + private final URI uri; + private boolean isAttempting; + private boolean isPrimary; public HealthStatusAndDiagnosticStringTuple(URI uri, HealthStatus status) { - this.diagnsoticString = uri.getPort() + ":" + status; + this.uri = uri; this.status = status; } + + public String getDiagnsoticString() { + return uri.getPort() + ":" + (isPrimary ? PRIMARY : SECONDARY) + ":" + + status + "|" + (isAttempting ? ATTEMPTING : NOT_ATTEMPTING); + } + + public void setAttempting(boolean attempting) { + isAttempting = attempting; + } + + public void setPrimary(boolean primary) { + isPrimary = primary; + } } } From 5e2ad3c8bef1e9370ab885564b2f16a9d6524118 Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Mon, 22 Apr 2024 16:53:11 -0700 Subject: [PATCH 02/16] Update sdk/cosmos/azure-cosmos/CHANGELOG.md --- sdk/cosmos/azure-cosmos/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index badfca11091d..e72a185ccc6d 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -3,7 +3,7 @@ ### 4.59.0-beta.1 (Unreleased) #### Features Added -* Added diagnostic fields for `quorumAckedLSN` and `currentReplicaSetSize`. Changed `replicaStatusList` to include all replicas and more information. - See [PR 39807]() +* Added diagnostic fields for `quorumAckedLSN` and `currentReplicaSetSize`. Changed `replicaStatusList` to include all replicas and more information. - See [PR 39844](https://github.com/Azure/azure-sdk-for-java/pull/39844) #### Breaking Changes From a5dceec51dbbbbbe4db971eb09a69acfc44d9d1d Mon Sep 17 00:00:00 2001 From: tvaron3 Date: Mon, 22 Apr 2024 21:41:43 -0700 Subject: [PATCH 03/16] changed tests to mock the new overload primary async method --- .../directconnectivity/AddressSelectorWrapper.java | 6 +++--- .../implementation/directconnectivity/StoreReaderTest.java | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java index 47dcf0645126..c8e62e6d708e 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java @@ -258,7 +258,7 @@ public AddressSelectorWrapper validate() { } public AddressSelectorWrapper verifyVesolvePrimaryUriAsyncCount(int count) { - Mockito.verify(addressSelector, Mockito.times(count)).resolvePrimaryUriAsync(Mockito.any(), Mockito.anyBoolean()); + Mockito.verify(addressSelector, Mockito.times(count)).resolvePrimaryUriAsync(Mockito.any(), Mockito.anyBoolean(), Mockito.anyMap()); return this; } @@ -314,7 +314,7 @@ public PrimaryReplicaMoveBuilder withPrimaryReplicaMove(Uri primaryURIBeforeForc } return Mono.just(primaryURIBeforeForceRefresh); - }).when(addressSelector).resolvePrimaryUriAsync(Mockito.any(RxDocumentServiceRequest.class), Mockito.anyBoolean()); + }).when(addressSelector).resolvePrimaryUriAsync(Mockito.any(RxDocumentServiceRequest.class), Mockito.anyBoolean(), Mockito.anyMap()); Mockito.doAnswer((invocation -> { capture(invocation); @@ -482,7 +482,7 @@ public AddressSelectorWrapper build() { Mockito.doAnswer((invocation) -> { capture(invocation); return Mono.just(primaryAddress); - }).when(addressSelector).resolvePrimaryUriAsync(Mockito.any(RxDocumentServiceRequest.class), Mockito.anyBoolean()); + }).when(addressSelector).resolvePrimaryUriAsync(Mockito.any(RxDocumentServiceRequest.class), Mockito.anyBoolean(), Mockito.anyMap()); Mockito.doAnswer((invocation -> { capture(invocation); diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java index ebccbd90c8b8..165bf3313727 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java @@ -498,7 +498,7 @@ public void readPrimaryAsync() { request.requestContext.requestChargeTracker = new RequestChargeTracker(); Mockito.doReturn(Mono.just(primaryURI)).when(addressSelector).resolvePrimaryUriAsync( - Mockito.eq(request) , Mockito.eq(false)); + Mockito.eq(request) , Mockito.eq(false), Mockito.anyMap()); StoreResponse storeResponse = Mockito.mock(StoreResponse.class); Mockito.doReturn(Mono.just(storeResponse)).when(transportClient).invokeResourceOperationAsync(Mockito.eq(primaryURI), Mockito.eq(request)); @@ -529,7 +529,7 @@ public void readPrimaryAsync_GoneFromReplica() { request.requestContext.requestChargeTracker = new RequestChargeTracker(); Mockito.doReturn(Mono.just(primaryURI)).when(addressSelector).resolvePrimaryUriAsync( - Mockito.eq(request) , Mockito.eq(false)); + Mockito.eq(request) , Mockito.eq(false), Mockito.anyMap()); Mockito.doReturn(Mono.error(ExceptionBuilder.create().asGoneException())).when(transportClient).invokeResourceOperationAsync(Mockito.eq(primaryURI), Mockito.eq(request)); StoreReader storeReader = new StoreReader(transportClient, addressSelector, sessionContainer); @@ -586,7 +586,7 @@ public void readPrimaryAsync_Error() { request.requestContext.requestChargeTracker = new RequestChargeTracker(); Mockito.doReturn(Mono.just(primaryURI)).when(addressSelector).resolvePrimaryUriAsync( - Mockito.eq(request) , Mockito.eq(false)); + Mockito.eq(request) , Mockito.eq(false), Mockito.anyMap()); StoreResponse storeResponse = Mockito.mock(StoreResponse.class); Mockito.doReturn(Mono.just(storeResponse)).when(transportClient).invokeResourceOperationAsync(Mockito.eq(primaryURI), Mockito.eq(request)); From 8fa93d998e0880cf79903ec6a17d0433cc4bf9fc Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Sun, 28 Apr 2024 19:48:50 -0700 Subject: [PATCH 04/16] reacting to comments --- .../azure/cosmos/CosmosDiagnosticsTest.java | 17 +++--- .../directconnectivity/StoreReaderTest.java | 6 +- sdk/cosmos/azure-cosmos/CHANGELOG.md | 2 +- .../com/azure/cosmos/CosmosException.java | 7 ++- .../ImplementationBridgeHelpers.java | 2 +- .../directconnectivity/AddressSelector.java | 7 +-- .../directconnectivity/ConsistencyWriter.java | 16 ++--- .../directconnectivity/StoreReader.java | 42 +++++++------ .../directconnectivity/StoreResponse.java | 9 +-- .../StoreResponseDiagnostics.java | 5 +- .../directconnectivity/Uri.java | 61 +++++-------------- 11 files changed, 75 insertions(+), 99 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java index 563631b78fd7..6e511dcf660f 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java @@ -26,6 +26,7 @@ import com.azure.cosmos.implementation.directconnectivity.GatewayAddressCache; import com.azure.cosmos.implementation.directconnectivity.GlobalAddressResolver; import com.azure.cosmos.implementation.directconnectivity.ReflectionUtils; +import com.azure.cosmos.implementation.directconnectivity.Uri; import com.azure.cosmos.implementation.directconnectivity.rntbd.RntbdChannelStatistics; import com.azure.cosmos.implementation.http.HttpClient; import com.azure.cosmos.implementation.http.HttpClientConfig; @@ -977,14 +978,13 @@ public void directDiagnosticsOnException() throws Exception { assertThat(responseStatisticsList.size()).isGreaterThan(0); JsonNode storeResult = responseStatisticsList.get(0).get("storeResult"); assertThat(storeResult).isNotNull(); + int currentReplicaSetSize = storeResult.get("currentReplicaSetSize").asInt(-1); + assertThat(currentReplicaSetSize).isGreaterThan(0); JsonNode replicaStatusList = storeResult.get("replicaStatusList"); - assertThat(replicaStatusList.isArray()).isTrue(); - assertThat(replicaStatusList.size()).isGreaterThan(0); + assertThat(replicaStatusList.isObject()).isTrue(); + assertThat(replicaStatusList.get(Uri.ATTEMPTING).size()).isEqualTo(currentReplicaSetSize); int quorumAcked = storeResult.get("quorumAckedLSN").asInt(-1); assertThat(quorumAcked).isGreaterThan(0); - int currentReplicaSetSize = storeResult.get("currentReplicaSetSize").asInt(-1); - assertThat(currentReplicaSetSize).isGreaterThan(0); - } } @@ -1230,9 +1230,10 @@ private void validateRntbdStatistics(CosmosDiagnostics cosmosDiagnostics, int replicaSetSize = storeResult.get("currentReplicaSetSize").asInt(-1); assertThat(replicaSetSize).isGreaterThan(0); JsonNode replicaStatusList = storeResult.get("replicaStatusList"); - assertThat(replicaStatusList.isArray()).isTrue(); - assertThat(replicaStatusList.size()).isEqualTo(replicaSetSize); - String replicaStatusTxt = replicaStatusList.get(0).asText(); + assertThat(replicaStatusList.isObject()).isTrue(); + int replicasNum = replicaStatusList.get(Uri.ATTEMPTING).size() + replicaStatusList.get(Uri.NOT_ATTEMPTING).size(); + assertThat(replicasNum).isEqualTo(replicaSetSize); + String replicaStatusTxt = replicaStatusList.get(Uri.ATTEMPTING).get(0).asText(); assertThat(replicaStatusTxt.contains("Primary")).isTrue(); assertThat(replicaStatusTxt.contains("Connected")).isTrue(); // validate serviceEndpointStatistics diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java index 165bf3313727..41cafce384ed 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java @@ -37,7 +37,9 @@ import reactor.core.publisher.Mono; import java.util.Arrays; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; @@ -636,13 +638,15 @@ public void canParseLongLsn() { .withGlobalCommittedLsn(bigLsn) .build(); + Map> replicaStatusList = new HashMap<>(); + replicaStatusList.put(Uri.ATTEMPTING, Arrays.asList(primaryURI.getHealthStatusDiagnosticString())); StoreResult result = storeReader.createStoreResult( storeResponse, null, false, false, null, - Arrays.asList(primaryURI.getHealthStatusDiagnosticString())); + replicaStatusList); assertThat(result.globalCommittedLSN).isEqualTo(bigLsn); assertThat(result.lsn).isEqualTo(bigLsn); } diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index 6ef66c10f0ee..674399d32034 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -3,7 +3,6 @@ ### 4.59.0-beta.1 (Unreleased) #### Features Added -* Added diagnostic fields for `quorumAckedLSN` and `currentReplicaSetSize`. Changed `replicaStatusList` to include all replicas and more information. - See [PR 39844](https://github.com/Azure/azure-sdk-for-java/pull/39844) * Added public APIs `getCustomeSerializer` and `setCustomSerializer` to allow customers to specify custom payload transformations or serialization settings. - See [PR 38997](https://github.com/Azure/azure-sdk-for-java/pull/38997) #### Breaking Changes @@ -12,6 +11,7 @@ #### Other Changes * Load Blackbird or Afterburner into the ObjectMapper depending upon Java version and presence of modules in classpath. Make Afterburner and Blackbird optional maven dependencies. See - [PR 39689](https://github.com/Azure/azure-sdk-for-java/pull/39689) +* Added diagnostic fields for `quorumAckedLSN` and `currentReplicaSetSize`. Changed `replicaStatusList` to include all replicas and more information. - See [PR 39844](https://github.com/Azure/azure-sdk-for-java/pull/39844) ### 4.58.0 (2024-04-16) #### Other Changes diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosException.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosException.java index b53db5fb5465..257cc884d4e8 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosException.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosException.java @@ -22,6 +22,7 @@ import java.time.Duration; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -139,7 +140,7 @@ public class CosmosException extends AzureException { /*** * All selectable replica status. */ - private final List replicaStatusList = new ArrayList<>(); + private final Map> replicaStatusList = new HashMap<>(); /** * Fault injection ruleId @@ -610,7 +611,7 @@ List getFaultInjectionEvaluationResults() { return this.faultInjectionEvaluationResults; } - List getReplicaStatusList() { + Map> getReplicaStatusList() { return this.replicaStatusList; } @@ -626,7 +627,7 @@ public CosmosException createCosmosException(int statusCode, Exception innerExce } @Override - public List getReplicaStatusList(CosmosException cosmosException) { + public Map> getReplicaStatusList(CosmosException cosmosException) { return cosmosException.getReplicaStatusList(); } 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 034026a007da..96647f002c1c 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 @@ -1408,7 +1408,7 @@ public static void setCosmosExceptionAccessor(final CosmosExceptionAccessor newA public interface CosmosExceptionAccessor { CosmosException createCosmosException(int statusCode, Exception innerException); - List getReplicaStatusList(CosmosException cosmosException); + Map> getReplicaStatusList(CosmosException cosmosException); CosmosException setRntbdChannelStatistics(CosmosException cosmosException, RntbdChannelStatistics rntbdChannelStatistics); RntbdChannelStatistics getRntbdChannelStatistics(CosmosException cosmosException); diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java index 4cd9e9c49306..0424fb87c853 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java @@ -32,12 +32,7 @@ public Mono> resolveAllUriAsync( boolean forceRefresh) { Mono> allReplicaAddressesObs = this.resolveAddressesAsync(request, forceRefresh); return allReplicaAddressesObs.map(allReplicaAddresses -> allReplicaAddresses.stream().filter(a -> includePrimary || !a.isPrimary()) - .map(a -> { - if (includePrimary && a.isPrimary()) { - a.getPhysicalUri().setHealthStatusTuplePrimary(true); - } - return a.getPhysicalUri(); - }).collect(Collectors.toList())); + .map(a -> a.getPhysicalUri()).collect(Collectors.toList())); } public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolean forceAddressRefresh) { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java index bea8b90027c1..2526c9786832 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java @@ -37,6 +37,7 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; @@ -169,7 +170,7 @@ Mono writePrivateAsync( Mono> replicaAddressesObs = this.addressSelector.resolveAddressesAsync(request, forceRefresh); AtomicReference primaryURI = new AtomicReference<>(); - AtomicReference> replicaStatusList = new AtomicReference<>(); + Map> replicaStatusList = new ConcurrentHashMap<>(); Map replicaStatuses = new ConcurrentHashMap<>(); return replicaAddressesObs.flatMap(replicaAddresses -> { @@ -204,10 +205,9 @@ Mono writePrivateAsync( } catch (Exception e) { return Mono.error(e); } - primaryUri.setHealthStatusTuplePrimary(true); - primaryUri.setHealthStatusTupleAttempting(true); - replicaStatuses.replace(primaryUri, primaryUri.getHealthStatusDiagnosticString()); - replicaStatusList.set(new ArrayList<>(replicaStatuses.values())); + replicaStatuses.remove(primaryUri); + replicaStatusList.put(Uri.NOT_ATTEMPTING, new ArrayList<>(replicaStatuses.values())); + replicaStatusList.put(Uri.ATTEMPTING,Arrays.asList(primaryUri.getHealthStatusDiagnosticString())); return this.transportClient.invokeResourceOperationAsync(primaryUri, request) .doOnError( @@ -230,7 +230,7 @@ Mono writePrivateAsync( false, false, primaryUri, - replicaStatusList.get()); + replicaStatusList); String value = ex != null ? ex .getResponseHeaders() @@ -267,7 +267,7 @@ Mono writePrivateAsync( false, false, primaryURI.get(), - replicaStatusList.get()); + replicaStatusList); return barrierForGlobalStrong(request, response); }) .doFinally(signalType -> { @@ -279,7 +279,7 @@ Mono writePrivateAsync( request, false, false, - replicaStatusList.get()); + replicaStatusList); }); } else { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java index ca420ec6c686..8be31ee7dbe5 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java @@ -34,6 +34,7 @@ import reactor.core.publisher.SignalType; import reactor.core.scheduler.Schedulers; +import java.lang.reflect.Array; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -148,7 +149,7 @@ private Flux toStoreResult( Pair, Uri> storeRespAndURI, ReadMode readMode, boolean requiresValidLsn, - List replicaStatusList) { + Map> replicaStatusList) { return storeRespAndURI.getLeft() .flatMap(storeResponse -> { @@ -219,20 +220,23 @@ private Flux> readFromReplicas(List resultCollect return Flux.error(new GoneException()); } List, Uri>> readStoreTasks = new ArrayList<>(); - Map replicaStatuses = new HashMap<>(); + Map replicaStatusesAttempting = new HashMap<>(); + Map replicaStatusesNotAttempting = new HashMap<>(); + resolveApiResults.forEach(uri -> { - replicaStatuses.put(uri, uri.getHealthStatusDiagnosticString()); + replicaStatusesNotAttempting.put(uri, uri.getHealthStatusDiagnosticString()); }); List addressRandomPermutation = AddressEnumerator.getTransportAddresses(entity, resolveApiResults); // The health status of the Uri will change as the time goes by // what we really want to track is the health status snapshot at this moment addressRandomPermutation.forEach(uri -> { - uri.setHealthStatusTupleAttempting(true); - replicaStatuses.replace(uri, uri.getHealthStatusDiagnosticString()); + replicaStatusesNotAttempting.remove(uri); + replicaStatusesAttempting.put(uri, uri.getHealthStatusDiagnosticString()); }); - // how to add primary - List replicaStatusList = new ArrayList<>(replicaStatuses.values()); + Map> replicaStatusList = new HashMap<>(); + replicaStatusList.put(Uri.ATTEMPTING ,new ArrayList<>(replicaStatusesAttempting.values())); + replicaStatusList.put(Uri.NOT_ATTEMPTING, new ArrayList<>(replicaStatusesNotAttempting.values())); int startIndex = 0; @@ -329,7 +333,7 @@ void createAndRecordStoreResultForCancelledRequest( RxDocumentServiceRequest request, boolean requiresValidLsn, boolean useLocalLSNBasedHeaders, - List replicaStatusList) { + Map> replicaStatusList) { // record the diagnostics for in-progress requests for (CosmosException cosmosException : request.requestContext.rntbdCancelledRequestMap.values()) { Uri storePhysicalAddress = @@ -575,7 +579,7 @@ private Mono readPrimaryInternalAsync( entity, entity.requestContext.forceRefreshAddressCache, replicaStatuses); - AtomicReference> replicaStatusList = new AtomicReference<>(); + Map> replicaStatusList = new ConcurrentHashMap<>(); AtomicReference primaryUriReference = new AtomicReference<>(null); @@ -596,9 +600,9 @@ private Mono readPrimaryInternalAsync( this.readFromStoreAsync( primaryUri, entity); - primaryUri.setHealthStatusTuplePrimary(true); - replicaStatuses.replace(primaryUri, primaryUri.getHealthStatusDiagnosticString()); - replicaStatusList.set(new ArrayList<>(replicaStatuses.values())); + replicaStatuses.remove(primaryUri); + replicaStatusList.put(Uri.NOT_ATTEMPTING, new ArrayList<>(replicaStatuses.values())); + replicaStatusList.put(Uri.ATTEMPTING, Arrays.asList(primaryUri.getHealthStatusDiagnosticString())); return storeResponseObsAndUri.getLeft().flatMap( storeResponse -> { @@ -611,7 +615,7 @@ private Mono readPrimaryInternalAsync( requiresValidLsn, true, storeResponse != null ? storeResponseObsAndUri.getRight() : null, - replicaStatusList.get()); + replicaStatusList); return Mono.just(storeResult); } catch (CosmosException e) { @@ -644,7 +648,7 @@ private Mono readPrimaryInternalAsync( requiresValidLsn, true, primaryUriReference.get(), - replicaStatusList.get()); + replicaStatusList); return Mono.just(storeResult); } catch (CosmosException e) { // RxJava1 doesn't allow throwing checked exception from Observable operators @@ -656,7 +660,7 @@ private Mono readPrimaryInternalAsync( return; } - this.createAndRecordStoreResultForCancelledRequest(entity, requiresValidLsn, true, replicaStatusList.get()); + this.createAndRecordStoreResultForCancelledRequest(entity, requiresValidLsn, true, replicaStatusList); }); return storeResultObs.map(storeResult -> { @@ -752,7 +756,7 @@ StoreResult createAndRecordStoreResult( boolean requiresValidLsn, boolean useLocalLSNBasedHeaders, Uri storePhysicalAddress, - List replicaStatusList) { + Map> replicaStatusList) { StoreResult storeResult = this.createStoreResult( @@ -784,7 +788,7 @@ StoreResult createStoreResult(StoreResponse storeResponse, boolean requiresValidLsn, boolean useLocalLSNBasedHeaders, Uri storePhysicalAddress, - List replicaStatusList) { + Map> replicaStatusList) { if (responseException == null) { String headerValue = null; @@ -798,7 +802,7 @@ StoreResult createStoreResult(StoreResponse storeResponse, long itemLSN = -1; if (replicaStatusList != null) { - storeResponse.getReplicaStatusList().addAll(replicaStatusList); + storeResponse.getReplicaStatusList().putAll(replicaStatusList); } if ((headerValue = storeResponse.getHeaderValue( @@ -907,7 +911,7 @@ StoreResult createStoreResult(StoreResponse storeResponse, .CosmosExceptionHelper .getCosmosExceptionAccessor() .getReplicaStatusList(cosmosException) - .addAll(replicaStatusList); + .putAll(replicaStatusList); } String headerValue = cosmosException.getResponseHeaders().get(useLocalLSNBasedHeaders ? WFConstants.BackendHeaders.QUORUM_ACKED_LOCAL_LSN : WFConstants.BackendHeaders.QUORUM_ACKED_LSN); 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 d697e42a7ca8..22802953db5a 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 @@ -19,6 +19,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkArgument; @@ -37,7 +38,7 @@ public class StoreResponse { private RntbdChannelStatistics channelStatistics; private int rntbdRequestLength; private int rntbdResponseLength; - private final List replicaStatusList; + private final Map> replicaStatusList; private String faultInjectionRuleId; private List faultInjectionRuleEvaluationResults; @@ -63,7 +64,7 @@ public StoreResponse( } this.status = status; - replicaStatusList = new ArrayList<>(); + replicaStatusList = new HashMap<>(); if (contentStream != null) { try { this.responsePayload = new JsonNodeStorePayload(contentStream, responsePayloadLength); @@ -97,7 +98,7 @@ private StoreResponse( } this.status = status; - replicaStatusList = new ArrayList<>(); + replicaStatusList = new HashMap<>(); this.responsePayload = responsePayload; } @@ -245,7 +246,7 @@ int getSubStatusCode() { return subStatusCode; } - public List getReplicaStatusList() { + public Map> getReplicaStatusList() { return this.replicaStatusList; } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponseDiagnostics.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponseDiagnostics.java index 8d175831df04..dc5d652cce48 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponseDiagnostics.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponseDiagnostics.java @@ -17,6 +17,7 @@ import org.slf4j.LoggerFactory; import java.util.List; +import java.util.Map; /** * This represents diagnostics from store response OR from cosmos exception @@ -40,7 +41,7 @@ public class StoreResponseDiagnostics { private final int rntbdResponseLength; private final String exceptionMessage; private final String exceptionResponseHeaders; - private final List replicaStatusList; + private final Map> replicaStatusList; private final String faultInjectionRuleId; private final List faultInjectionEvaluationResults; @@ -196,5 +197,5 @@ public List getFaultInjectionEvaluationResults() { return this.faultInjectionEvaluationResults; } - public List getReplicaStatusList() { return this.replicaStatusList; } + public Map> getReplicaStatusList() { return this.replicaStatusList; } } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java index d4d3cd39033b..3eb51ec172b7 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java @@ -3,7 +3,6 @@ package com.azure.cosmos.implementation.directconnectivity; -import com.azure.cosmos.implementation.cpu.CpuMemoryReader; import com.azure.cosmos.implementation.directconnectivity.addressEnumerator.AddressEnumerator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -25,6 +24,10 @@ public class Uri { private volatile Instant lastUnhealthyPendingTimestamp; private volatile Instant lastTransitionToUnhealthyTimestamp; private volatile boolean isPrimary; + public static final String ATTEMPTING = "Attempting"; + public static final String NOT_ATTEMPTING = "NotAttempting"; + private static final String PRIMARY = "Primary"; + private static final String SECONDARY = "Secondary"; public static Uri create(String uriAsString) { return new Uri(uriAsString); @@ -55,12 +58,12 @@ public String getURIAsString() { return this.uriAsString; } - /*** - * Attention: This is only used for fault injection to easier decide whether the address is primary address. - * @param primary - */ public void setPrimary(boolean primary) { isPrimary = primary; + this.healthStatusTuple.updateAndGet(previousStatusTuple -> { + previousStatusTuple.diagnosticString = previousStatusTuple.diagnosticString.replace(SECONDARY, PRIMARY); + return previousStatusTuple; + }); } /*** @@ -140,7 +143,7 @@ public void setHealthStatus(HealthStatus status) { status, previousStatusTuple, newStatus); } - return new HealthStatusAndDiagnosticStringTuple(this.uri, newStatus); + return new HealthStatusAndDiagnosticStringTuple(this.uri, newStatus, isPrimary); }); } @@ -184,25 +187,7 @@ public boolean shouldRefreshHealthStatus() { } public String getHealthStatusDiagnosticString() { - return this.healthStatusTuple.get().getDiagnsoticString(); - } - - public void setHealthStatusTupleAttempting(boolean attempting) { - this.healthStatusTuple.updateAndGet(previousStatusTuple -> - { - previousStatusTuple.setAttempting(attempting); - return previousStatusTuple; - } - ); - } - - public void setHealthStatusTuplePrimary(boolean primary) { - this.healthStatusTuple.updateAndGet(previousStatusTuple -> - { - previousStatusTuple.setPrimary(primary); - return previousStatusTuple; - } - ); + return this.healthStatusTuple.get().diagnosticString; } @Override @@ -249,31 +234,15 @@ public int getPriority() { } static class HealthStatusAndDiagnosticStringTuple { + private String diagnosticString; private final HealthStatus status; - private final String ATTEMPTING = "Attempting"; - private final String NOT_ATTEMPTING = "NotAttempting"; - private final String PRIMARY = "Primary"; - private final String SECONDARY = "Secondary"; - private final URI uri; - private boolean isAttempting; - private boolean isPrimary; - public HealthStatusAndDiagnosticStringTuple(URI uri, HealthStatus status) { - this.uri = uri; + this.diagnosticString = uri.getPort() + ":" + Uri.SECONDARY + ":" + status; this.status = status; } - - public String getDiagnsoticString() { - return uri.getPort() + ":" + (isPrimary ? PRIMARY : SECONDARY) + ":" + - status + "|" + (isAttempting ? ATTEMPTING : NOT_ATTEMPTING); - } - - public void setAttempting(boolean attempting) { - isAttempting = attempting; - } - - public void setPrimary(boolean primary) { - isPrimary = primary; + public HealthStatusAndDiagnosticStringTuple(URI uri, HealthStatus status, boolean isPrimary) { + this.diagnosticString = uri.getPort() + ":" + (isPrimary ? Uri.PRIMARY : Uri.SECONDARY) + ":" + status; + this.status = status; } } } From 285a996d0831963190c247096a306efe41ebb24a Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Mon, 29 Apr 2024 10:41:23 -0700 Subject: [PATCH 05/16] reacting to comments --- .../implementation/directconnectivity/Uri.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java index 3eb51ec172b7..decad270278c 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java @@ -60,10 +60,12 @@ public String getURIAsString() { public void setPrimary(boolean primary) { isPrimary = primary; - this.healthStatusTuple.updateAndGet(previousStatusTuple -> { - previousStatusTuple.diagnosticString = previousStatusTuple.diagnosticString.replace(SECONDARY, PRIMARY); - return previousStatusTuple; - }); + if (primary) { + this.healthStatusTuple.updateAndGet(previousStatusTuple -> { + previousStatusTuple.diagnosticString = previousStatusTuple.diagnosticString.replace(SECONDARY, PRIMARY); + return previousStatusTuple; + }); + } } /*** @@ -143,7 +145,7 @@ public void setHealthStatus(HealthStatus status) { status, previousStatusTuple, newStatus); } - return new HealthStatusAndDiagnosticStringTuple(this.uri, newStatus, isPrimary); + return new HealthStatusAndDiagnosticStringTuple(this.uri, newStatus, this.isPrimary); }); } From e399d75c950801d50cc509188a047018facd2dea Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Thu, 2 May 2024 13:24:31 -0700 Subject: [PATCH 06/16] reacting to comments --- .../azure/cosmos/CosmosDiagnosticsTest.java | 2 +- .../AddressSelectorWrapper.java | 6 ++--- .../directconnectivity/StoreReaderTest.java | 6 ++--- .../directconnectivity/AddressSelector.java | 8 +++---- .../directconnectivity/ConsistencyWriter.java | 12 +++++----- .../directconnectivity/StoreReader.java | 23 ++++++++----------- .../directconnectivity/Uri.java | 6 ++--- 7 files changed, 30 insertions(+), 33 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java index 6e511dcf660f..f61dea836095 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java @@ -1231,7 +1231,7 @@ private void validateRntbdStatistics(CosmosDiagnostics cosmosDiagnostics, assertThat(replicaSetSize).isGreaterThan(0); JsonNode replicaStatusList = storeResult.get("replicaStatusList"); assertThat(replicaStatusList.isObject()).isTrue(); - int replicasNum = replicaStatusList.get(Uri.ATTEMPTING).size() + replicaStatusList.get(Uri.NOT_ATTEMPTING).size(); + int replicasNum = replicaStatusList.get(Uri.ATTEMPTING).size() + replicaStatusList.get(Uri.IGNORING).size(); assertThat(replicasNum).isEqualTo(replicaSetSize); String replicaStatusTxt = replicaStatusList.get(Uri.ATTEMPTING).get(0).asText(); assertThat(replicaStatusTxt.contains("Primary")).isTrue(); diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java index c8e62e6d708e..0c38b904dab4 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java @@ -258,7 +258,7 @@ public AddressSelectorWrapper validate() { } public AddressSelectorWrapper verifyVesolvePrimaryUriAsyncCount(int count) { - Mockito.verify(addressSelector, Mockito.times(count)).resolvePrimaryUriAsync(Mockito.any(), Mockito.anyBoolean(), Mockito.anyMap()); + Mockito.verify(addressSelector, Mockito.times(count)).resolvePrimaryUriAsync(Mockito.any(), Mockito.anyBoolean(), Mockito.any()); return this; } @@ -314,7 +314,7 @@ public PrimaryReplicaMoveBuilder withPrimaryReplicaMove(Uri primaryURIBeforeForc } return Mono.just(primaryURIBeforeForceRefresh); - }).when(addressSelector).resolvePrimaryUriAsync(Mockito.any(RxDocumentServiceRequest.class), Mockito.anyBoolean(), Mockito.anyMap()); + }).when(addressSelector).resolvePrimaryUriAsync(Mockito.any(RxDocumentServiceRequest.class), Mockito.anyBoolean(), Mockito.any()); Mockito.doAnswer((invocation -> { capture(invocation); @@ -482,7 +482,7 @@ public AddressSelectorWrapper build() { Mockito.doAnswer((invocation) -> { capture(invocation); return Mono.just(primaryAddress); - }).when(addressSelector).resolvePrimaryUriAsync(Mockito.any(RxDocumentServiceRequest.class), Mockito.anyBoolean(), Mockito.anyMap()); + }).when(addressSelector).resolvePrimaryUriAsync(Mockito.any(RxDocumentServiceRequest.class), Mockito.anyBoolean(), Mockito.any()); Mockito.doAnswer((invocation -> { capture(invocation); diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java index 41cafce384ed..c5ae10d71aee 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java @@ -500,7 +500,7 @@ public void readPrimaryAsync() { request.requestContext.requestChargeTracker = new RequestChargeTracker(); Mockito.doReturn(Mono.just(primaryURI)).when(addressSelector).resolvePrimaryUriAsync( - Mockito.eq(request) , Mockito.eq(false), Mockito.anyMap()); + Mockito.eq(request) , Mockito.eq(false), Mockito.any()); StoreResponse storeResponse = Mockito.mock(StoreResponse.class); Mockito.doReturn(Mono.just(storeResponse)).when(transportClient).invokeResourceOperationAsync(Mockito.eq(primaryURI), Mockito.eq(request)); @@ -531,7 +531,7 @@ public void readPrimaryAsync_GoneFromReplica() { request.requestContext.requestChargeTracker = new RequestChargeTracker(); Mockito.doReturn(Mono.just(primaryURI)).when(addressSelector).resolvePrimaryUriAsync( - Mockito.eq(request) , Mockito.eq(false), Mockito.anyMap()); + Mockito.eq(request) , Mockito.eq(false), Mockito.any()); Mockito.doReturn(Mono.error(ExceptionBuilder.create().asGoneException())).when(transportClient).invokeResourceOperationAsync(Mockito.eq(primaryURI), Mockito.eq(request)); StoreReader storeReader = new StoreReader(transportClient, addressSelector, sessionContainer); @@ -588,7 +588,7 @@ public void readPrimaryAsync_Error() { request.requestContext.requestChargeTracker = new RequestChargeTracker(); Mockito.doReturn(Mono.just(primaryURI)).when(addressSelector).resolvePrimaryUriAsync( - Mockito.eq(request) , Mockito.eq(false), Mockito.anyMap()); + Mockito.eq(request) , Mockito.eq(false), Mockito.any()); StoreResponse storeResponse = Mockito.mock(StoreResponse.class); Mockito.doReturn(Mono.just(storeResponse)).when(transportClient).invokeResourceOperationAsync(Mockito.eq(primaryURI), Mockito.eq(request)); diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java index 0424fb87c853..ec1d94fa29f6 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java @@ -15,6 +15,7 @@ import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; public class AddressSelector { @@ -46,13 +47,12 @@ public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolea }); } - public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolean forceAddressRefresh, Map replicaStatuses) { + public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolean forceAddressRefresh, AtomicReference> replicaStatuses) { Mono> replicaAddressesObs = this.resolveAddressesAsync(request, forceAddressRefresh); return replicaAddressesObs.flatMap(replicaAddresses -> { try { - replicaAddresses.forEach(replica -> { - replicaStatuses.put(replica.getPhysicalUri(), replica.getPhysicalUri().getHealthStatusDiagnosticString()); - }); + replicaAddresses.stream().filter(replica -> !replica.isPrimary()).forEach(replica -> + replicaStatuses.get().add(replica.getPhysicalUri().getHealthStatusDiagnosticString())); return Mono.just(AddressSelector.getPrimaryUri(request, replicaAddresses)); } catch (Exception e) { return Mono.error(e); diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java index 2526c9786832..7e81de2323c9 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java @@ -37,8 +37,6 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; -import java.util.HashMap; -import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -171,7 +169,8 @@ Mono writePrivateAsync( Mono> replicaAddressesObs = this.addressSelector.resolveAddressesAsync(request, forceRefresh); AtomicReference primaryURI = new AtomicReference<>(); Map> replicaStatusList = new ConcurrentHashMap<>(); - Map replicaStatuses = new ConcurrentHashMap<>(); + AtomicReference> replicaStatuses = new AtomicReference<>(); + replicaStatuses.set(new ArrayList<>()); return replicaAddressesObs.flatMap(replicaAddresses -> { try { @@ -179,7 +178,9 @@ Mono writePrivateAsync( replicaAddresses.forEach(replicaAddress -> { Uri uri = replicaAddress.getPhysicalUri(); contactedReplicas.add(uri.getURI()); - replicaStatuses.put(uri, uri.getHealthStatusDiagnosticString()); + if (!uri.isPrimary()) { + replicaStatuses.get().add(uri.getHealthStatusDiagnosticString()); + } }); BridgeInternal.setContactedReplicas(request.requestContext.cosmosDiagnostics, contactedReplicas); return Mono.just(AddressSelector.getPrimaryUri(request, replicaAddresses)); @@ -205,8 +206,7 @@ Mono writePrivateAsync( } catch (Exception e) { return Mono.error(e); } - replicaStatuses.remove(primaryUri); - replicaStatusList.put(Uri.NOT_ATTEMPTING, new ArrayList<>(replicaStatuses.values())); + replicaStatusList.put(Uri.IGNORING, replicaStatuses.get()); replicaStatusList.put(Uri.ATTEMPTING,Arrays.asList(primaryUri.getHealthStatusDiagnosticString())); return this.transportClient.invokeResourceOperationAsync(primaryUri, request) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java index 8be31ee7dbe5..69bd91ddec1f 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java @@ -34,7 +34,6 @@ import reactor.core.publisher.SignalType; import reactor.core.scheduler.Schedulers; -import java.lang.reflect.Array; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -220,23 +219,22 @@ private Flux> readFromReplicas(List resultCollect return Flux.error(new GoneException()); } List, Uri>> readStoreTasks = new ArrayList<>(); - Map replicaStatusesAttempting = new HashMap<>(); - Map replicaStatusesNotAttempting = new HashMap<>(); + List replicaStatusesAttempting = new ArrayList<>(); + List replicaStatusesNotAttempting = new ArrayList<>(); - resolveApiResults.forEach(uri -> { - replicaStatusesNotAttempting.put(uri, uri.getHealthStatusDiagnosticString()); - }); List addressRandomPermutation = AddressEnumerator.getTransportAddresses(entity, resolveApiResults); // The health status of the Uri will change as the time goes by // what we really want to track is the health status snapshot at this moment addressRandomPermutation.forEach(uri -> { - replicaStatusesNotAttempting.remove(uri); - replicaStatusesAttempting.put(uri, uri.getHealthStatusDiagnosticString()); + replicaStatusesAttempting.add(uri.getHealthStatusDiagnosticString()); }); Map> replicaStatusList = new HashMap<>(); - replicaStatusList.put(Uri.ATTEMPTING ,new ArrayList<>(replicaStatusesAttempting.values())); - replicaStatusList.put(Uri.NOT_ATTEMPTING, new ArrayList<>(replicaStatusesNotAttempting.values())); + resolveApiResults.stream().filter(uri -> !replicaStatusesAttempting.contains(uri.getHealthStatusDiagnosticString())).forEach(uri -> + replicaStatusesNotAttempting.add(uri.getHealthStatusDiagnosticString()) + ); + replicaStatusList.put(Uri.ATTEMPTING , replicaStatusesAttempting); + replicaStatusList.put(Uri.IGNORING, replicaStatusesNotAttempting); int startIndex = 0; @@ -574,7 +572,7 @@ private Mono readPrimaryInternalAsync( return Mono.error(new GoneException()); } - Map replicaStatuses = new ConcurrentHashMap<>(); + AtomicReference> replicaStatuses = new AtomicReference<>(); Mono primaryUriObs = this.addressSelector.resolvePrimaryUriAsync( entity, entity.requestContext.forceRefreshAddressCache, replicaStatuses); @@ -600,8 +598,7 @@ private Mono readPrimaryInternalAsync( this.readFromStoreAsync( primaryUri, entity); - replicaStatuses.remove(primaryUri); - replicaStatusList.put(Uri.NOT_ATTEMPTING, new ArrayList<>(replicaStatuses.values())); + replicaStatusList.put(Uri.IGNORING, replicaStatuses.get()); replicaStatusList.put(Uri.ATTEMPTING, Arrays.asList(primaryUri.getHealthStatusDiagnosticString())); return storeResponseObsAndUri.getLeft().flatMap( diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java index decad270278c..7b41bffd3f77 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java @@ -25,9 +25,9 @@ public class Uri { private volatile Instant lastTransitionToUnhealthyTimestamp; private volatile boolean isPrimary; public static final String ATTEMPTING = "Attempting"; - public static final String NOT_ATTEMPTING = "NotAttempting"; - private static final String PRIMARY = "Primary"; - private static final String SECONDARY = "Secondary"; + public static final String IGNORING = "Ignoring"; + private static final String PRIMARY = "P"; + private static final String SECONDARY = "S"; public static Uri create(String uriAsString) { return new Uri(uriAsString); From e221a20190f854f3c48466f85ec15ffe363abe24 Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Fri, 3 May 2024 17:01:12 -0700 Subject: [PATCH 07/16] reacting to comment --- .../directconnectivity/StoreReaderTest.java | 5 ++-- .../com/azure/cosmos/CosmosException.java | 7 ++--- .../ImplementationBridgeHelpers.java | 3 ++- .../directconnectivity/AddressSelector.java | 5 ++-- .../directconnectivity/ConsistencyWriter.java | 13 +++++----- .../directconnectivity/StoreReader.java | 26 ++++++++++--------- .../directconnectivity/StoreResponse.java | 5 ++-- .../StoreResponseDiagnostics.java | 5 ++-- .../directconnectivity/Uri.java | 3 +-- 9 files changed, 40 insertions(+), 32 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java index c5ae10d71aee..a3fc041c502c 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java @@ -40,6 +40,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; @@ -638,8 +639,8 @@ public void canParseLongLsn() { .withGlobalCommittedLsn(bigLsn) .build(); - Map> replicaStatusList = new HashMap<>(); - replicaStatusList.put(Uri.ATTEMPTING, Arrays.asList(primaryURI.getHealthStatusDiagnosticString())); + Map> replicaStatusList = new HashMap<>(); + replicaStatusList.put(Uri.ATTEMPTING, Set.of(primaryURI.getHealthStatusDiagnosticString())); StoreResult result = storeReader.createStoreResult( storeResponse, null, diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosException.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosException.java index 257cc884d4e8..0cf4870cf485 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosException.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosException.java @@ -25,6 +25,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; @@ -140,7 +141,7 @@ public class CosmosException extends AzureException { /*** * All selectable replica status. */ - private final Map> replicaStatusList = new HashMap<>(); + private final Map> replicaStatusList = new HashMap<>(); /** * Fault injection ruleId @@ -611,7 +612,7 @@ List getFaultInjectionEvaluationResults() { return this.faultInjectionEvaluationResults; } - Map> getReplicaStatusList() { + Map> getReplicaStatusList() { return this.replicaStatusList; } @@ -627,7 +628,7 @@ public CosmosException createCosmosException(int statusCode, Exception innerExce } @Override - public Map> getReplicaStatusList(CosmosException cosmosException) { + public Map> getReplicaStatusList(CosmosException cosmosException) { return cosmosException.getReplicaStatusList(); } 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 96647f002c1c..a5b2a4986145 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 @@ -84,6 +84,7 @@ import java.util.EnumSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicBoolean; @@ -1408,7 +1409,7 @@ public static void setCosmosExceptionAccessor(final CosmosExceptionAccessor newA public interface CosmosExceptionAccessor { CosmosException createCosmosException(int statusCode, Exception innerException); - Map> getReplicaStatusList(CosmosException cosmosException); + Map> getReplicaStatusList(CosmosException cosmosException); CosmosException setRntbdChannelStatistics(CosmosException cosmosException, RntbdChannelStatistics rntbdChannelStatistics); RntbdChannelStatistics getRntbdChannelStatistics(CosmosException cosmosException); diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java index ec1d94fa29f6..e3cde37f96c0 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java @@ -15,6 +15,7 @@ import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; @@ -47,12 +48,12 @@ public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolea }); } - public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolean forceAddressRefresh, AtomicReference> replicaStatuses) { + public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolean forceAddressRefresh, Set replicaStatuses) { Mono> replicaAddressesObs = this.resolveAddressesAsync(request, forceAddressRefresh); return replicaAddressesObs.flatMap(replicaAddresses -> { try { replicaAddresses.stream().filter(replica -> !replica.isPrimary()).forEach(replica -> - replicaStatuses.get().add(replica.getPhysicalUri().getHealthStatusDiagnosticString())); + replicaStatuses.add(replica.getPhysicalUri().getHealthStatusDiagnosticString())); return Mono.just(AddressSelector.getPrimaryUri(request, replicaAddresses)); } catch (Exception e) { return Mono.error(e); diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java index 7e81de2323c9..f7ddd5a25601 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java @@ -37,8 +37,10 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -168,9 +170,8 @@ Mono writePrivateAsync( Mono> replicaAddressesObs = this.addressSelector.resolveAddressesAsync(request, forceRefresh); AtomicReference primaryURI = new AtomicReference<>(); - Map> replicaStatusList = new ConcurrentHashMap<>(); - AtomicReference> replicaStatuses = new AtomicReference<>(); - replicaStatuses.set(new ArrayList<>()); + Map> replicaStatusList = new ConcurrentHashMap<>(); + Set replicaStatuses = Collections.newSetFromMap(new ConcurrentHashMap<>()); return replicaAddressesObs.flatMap(replicaAddresses -> { try { @@ -179,7 +180,7 @@ Mono writePrivateAsync( Uri uri = replicaAddress.getPhysicalUri(); contactedReplicas.add(uri.getURI()); if (!uri.isPrimary()) { - replicaStatuses.get().add(uri.getHealthStatusDiagnosticString()); + replicaStatuses.add(uri.getHealthStatusDiagnosticString()); } }); BridgeInternal.setContactedReplicas(request.requestContext.cosmosDiagnostics, contactedReplicas); @@ -206,8 +207,8 @@ Mono writePrivateAsync( } catch (Exception e) { return Mono.error(e); } - replicaStatusList.put(Uri.IGNORING, replicaStatuses.get()); - replicaStatusList.put(Uri.ATTEMPTING,Arrays.asList(primaryUri.getHealthStatusDiagnosticString())); + replicaStatusList.put(Uri.IGNORING, replicaStatuses); + replicaStatusList.put(Uri.ATTEMPTING,Set.of(primaryUri.getHealthStatusDiagnosticString())); return this.transportClient.invokeResourceOperationAsync(primaryUri, request) .doOnError( diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java index 69bd91ddec1f..eb51fbfabf0b 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java @@ -38,8 +38,10 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -148,7 +150,7 @@ private Flux toStoreResult( Pair, Uri> storeRespAndURI, ReadMode readMode, boolean requiresValidLsn, - Map> replicaStatusList) { + Map> replicaStatusList) { return storeRespAndURI.getLeft() .flatMap(storeResponse -> { @@ -219,9 +221,8 @@ private Flux> readFromReplicas(List resultCollect return Flux.error(new GoneException()); } List, Uri>> readStoreTasks = new ArrayList<>(); - List replicaStatusesAttempting = new ArrayList<>(); - List replicaStatusesNotAttempting = new ArrayList<>(); - + Set replicaStatusesAttempting = new HashSet<>(); + Set replicaStatusesNotAttempting = new HashSet<>(); List addressRandomPermutation = AddressEnumerator.getTransportAddresses(entity, resolveApiResults); // The health status of the Uri will change as the time goes by @@ -229,7 +230,7 @@ private Flux> readFromReplicas(List resultCollect addressRandomPermutation.forEach(uri -> { replicaStatusesAttempting.add(uri.getHealthStatusDiagnosticString()); }); - Map> replicaStatusList = new HashMap<>(); + Map> replicaStatusList = new HashMap<>(); resolveApiResults.stream().filter(uri -> !replicaStatusesAttempting.contains(uri.getHealthStatusDiagnosticString())).forEach(uri -> replicaStatusesNotAttempting.add(uri.getHealthStatusDiagnosticString()) ); @@ -331,7 +332,7 @@ void createAndRecordStoreResultForCancelledRequest( RxDocumentServiceRequest request, boolean requiresValidLsn, boolean useLocalLSNBasedHeaders, - Map> replicaStatusList) { + Map> replicaStatusList) { // record the diagnostics for in-progress requests for (CosmosException cosmosException : request.requestContext.rntbdCancelledRequestMap.values()) { Uri storePhysicalAddress = @@ -572,12 +573,12 @@ private Mono readPrimaryInternalAsync( return Mono.error(new GoneException()); } - AtomicReference> replicaStatuses = new AtomicReference<>(); + Set replicaStatuses = Collections.newSetFromMap(new ConcurrentHashMap<>()); Mono primaryUriObs = this.addressSelector.resolvePrimaryUriAsync( entity, entity.requestContext.forceRefreshAddressCache, replicaStatuses); - Map> replicaStatusList = new ConcurrentHashMap<>(); + Map> replicaStatusList = new ConcurrentHashMap<>(); AtomicReference primaryUriReference = new AtomicReference<>(null); @@ -598,8 +599,9 @@ private Mono readPrimaryInternalAsync( this.readFromStoreAsync( primaryUri, entity); - replicaStatusList.put(Uri.IGNORING, replicaStatuses.get()); - replicaStatusList.put(Uri.ATTEMPTING, Arrays.asList(primaryUri.getHealthStatusDiagnosticString())); + + replicaStatusList.put(Uri.IGNORING, replicaStatuses); + replicaStatusList.put(Uri.ATTEMPTING, Set.of(primaryUri.getHealthStatusDiagnosticString())); return storeResponseObsAndUri.getLeft().flatMap( storeResponse -> { @@ -753,7 +755,7 @@ StoreResult createAndRecordStoreResult( boolean requiresValidLsn, boolean useLocalLSNBasedHeaders, Uri storePhysicalAddress, - Map> replicaStatusList) { + Map> replicaStatusList) { StoreResult storeResult = this.createStoreResult( @@ -785,7 +787,7 @@ StoreResult createStoreResult(StoreResponse storeResponse, boolean requiresValidLsn, boolean useLocalLSNBasedHeaders, Uri storePhysicalAddress, - Map> replicaStatusList) { + Map> replicaStatusList) { if (responseException == null) { String headerValue = null; 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 22802953db5a..4b1872c39b41 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 @@ -19,6 +19,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkArgument; @@ -38,7 +39,7 @@ public class StoreResponse { private RntbdChannelStatistics channelStatistics; private int rntbdRequestLength; private int rntbdResponseLength; - private final Map> replicaStatusList; + private final Map> replicaStatusList; private String faultInjectionRuleId; private List faultInjectionRuleEvaluationResults; @@ -246,7 +247,7 @@ int getSubStatusCode() { return subStatusCode; } - public Map> getReplicaStatusList() { + public Map> getReplicaStatusList() { return this.replicaStatusList; } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponseDiagnostics.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponseDiagnostics.java index dc5d652cce48..101b280c45e4 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponseDiagnostics.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponseDiagnostics.java @@ -18,6 +18,7 @@ import java.util.List; import java.util.Map; +import java.util.Set; /** * This represents diagnostics from store response OR from cosmos exception @@ -41,7 +42,7 @@ public class StoreResponseDiagnostics { private final int rntbdResponseLength; private final String exceptionMessage; private final String exceptionResponseHeaders; - private final Map> replicaStatusList; + private final Map> replicaStatusList; private final String faultInjectionRuleId; private final List faultInjectionEvaluationResults; @@ -197,5 +198,5 @@ public List getFaultInjectionEvaluationResults() { return this.faultInjectionEvaluationResults; } - public Map> getReplicaStatusList() { return this.replicaStatusList; } + public Map> getReplicaStatusList() { return this.replicaStatusList; } } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java index 7b41bffd3f77..bbc605eaa610 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java @@ -239,8 +239,7 @@ static class HealthStatusAndDiagnosticStringTuple { private String diagnosticString; private final HealthStatus status; public HealthStatusAndDiagnosticStringTuple(URI uri, HealthStatus status) { - this.diagnosticString = uri.getPort() + ":" + Uri.SECONDARY + ":" + status; - this.status = status; + this(uri, status, false); } public HealthStatusAndDiagnosticStringTuple(URI uri, HealthStatus status, boolean isPrimary) { this.diagnosticString = uri.getPort() + ":" + (isPrimary ? Uri.PRIMARY : Uri.SECONDARY) + ":" + status; From b7d21c0049fd4a3843c54d653fcce36ce139869c Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Sat, 4 May 2024 23:38:47 -0700 Subject: [PATCH 08/16] removed set.of cause used another package --- .../implementation/directconnectivity/ConsistencyWriter.java | 3 ++- .../cosmos/implementation/directconnectivity/StoreReader.java | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java index f7ddd5a25601..8ba1ab0f94ac 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java @@ -38,6 +38,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -208,7 +209,7 @@ Mono writePrivateAsync( return Mono.error(e); } replicaStatusList.put(Uri.IGNORING, replicaStatuses); - replicaStatusList.put(Uri.ATTEMPTING,Set.of(primaryUri.getHealthStatusDiagnosticString())); + replicaStatusList.put(Uri.ATTEMPTING,new HashSet<>(Arrays.asList(primaryUri.getHealthStatusDiagnosticString()))); return this.transportClient.invokeResourceOperationAsync(primaryUri, request) .doOnError( diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java index eb51fbfabf0b..542c00b56867 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java @@ -601,7 +601,7 @@ private Mono readPrimaryInternalAsync( entity); replicaStatusList.put(Uri.IGNORING, replicaStatuses); - replicaStatusList.put(Uri.ATTEMPTING, Set.of(primaryUri.getHealthStatusDiagnosticString())); + replicaStatusList.put(Uri.ATTEMPTING, new HashSet<>(Arrays.asList(primaryUri.getHealthStatusDiagnosticString()))); return storeResponseObsAndUri.getLeft().flatMap( storeResponse -> { From 879779caf42195ece0b9de8baddf44afd5fd4146 Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Sat, 4 May 2024 23:38:47 -0700 Subject: [PATCH 09/16] removed set.of cause used another package --- .../implementation/directconnectivity/StoreReaderTest.java | 3 ++- .../implementation/directconnectivity/ConsistencyWriter.java | 3 ++- .../cosmos/implementation/directconnectivity/StoreReader.java | 2 +- 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java index a3fc041c502c..710946c2325c 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/StoreReaderTest.java @@ -38,6 +38,7 @@ import java.util.Arrays; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -640,7 +641,7 @@ public void canParseLongLsn() { .build(); Map> replicaStatusList = new HashMap<>(); - replicaStatusList.put(Uri.ATTEMPTING, Set.of(primaryURI.getHealthStatusDiagnosticString())); + replicaStatusList.put(Uri.ATTEMPTING, new HashSet<>(Arrays.asList(primaryURI.getHealthStatusDiagnosticString()))); StoreResult result = storeReader.createStoreResult( storeResponse, null, diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java index f7ddd5a25601..8ba1ab0f94ac 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java @@ -38,6 +38,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -208,7 +209,7 @@ Mono writePrivateAsync( return Mono.error(e); } replicaStatusList.put(Uri.IGNORING, replicaStatuses); - replicaStatusList.put(Uri.ATTEMPTING,Set.of(primaryUri.getHealthStatusDiagnosticString())); + replicaStatusList.put(Uri.ATTEMPTING,new HashSet<>(Arrays.asList(primaryUri.getHealthStatusDiagnosticString()))); return this.transportClient.invokeResourceOperationAsync(primaryUri, request) .doOnError( diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java index eb51fbfabf0b..542c00b56867 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java @@ -601,7 +601,7 @@ private Mono readPrimaryInternalAsync( entity); replicaStatusList.put(Uri.IGNORING, replicaStatuses); - replicaStatusList.put(Uri.ATTEMPTING, Set.of(primaryUri.getHealthStatusDiagnosticString())); + replicaStatusList.put(Uri.ATTEMPTING, new HashSet<>(Arrays.asList(primaryUri.getHealthStatusDiagnosticString()))); return storeResponseObsAndUri.getLeft().flatMap( storeResponse -> { From d24f674098505366696fc37e27c6720527c7d93b Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Mon, 3 Jun 2024 10:24:24 -0700 Subject: [PATCH 10/16] Reacted to comments --- .../azure/cosmos/CosmosDiagnosticsTest.java | 7 +++--- .../directconnectivity/AddressSelector.java | 24 +++++++++++++++++++ .../directconnectivity/ConsistencyWriter.java | 2 +- .../directconnectivity/StoreReader.java | 16 ++++++++----- .../directconnectivity/StoreResponse.java | 2 -- .../directconnectivity/Uri.java | 6 ++--- 6 files changed, 40 insertions(+), 17 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java index 6978dd84e64c..f692a2433031 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java @@ -984,12 +984,11 @@ public void directDiagnosticsOnException() throws Exception { JsonNode storeResult = responseStatisticsList.get(0).get("storeResult"); assertThat(storeResult).isNotNull(); int currentReplicaSetSize = storeResult.get("currentReplicaSetSize").asInt(-1); - assertThat(currentReplicaSetSize).isGreaterThan(0); + assertThat(currentReplicaSetSize).isEqualTo(-1); JsonNode replicaStatusList = storeResult.get("replicaStatusList"); assertThat(replicaStatusList.isObject()).isTrue(); - assertThat(replicaStatusList.get(Uri.ATTEMPTING).size()).isEqualTo(currentReplicaSetSize); int quorumAcked = storeResult.get("quorumAckedLSN").asInt(-1); - assertThat(quorumAcked).isGreaterThan(0); + assertThat(quorumAcked).isEqualTo(-1); } } @@ -1239,7 +1238,7 @@ private void validateRntbdStatistics(CosmosDiagnostics cosmosDiagnostics, int replicasNum = replicaStatusList.get(Uri.ATTEMPTING).size() + replicaStatusList.get(Uri.IGNORING).size(); assertThat(replicasNum).isEqualTo(replicaSetSize); String replicaStatusTxt = replicaStatusList.get(Uri.ATTEMPTING).get(0).asText(); - assertThat(replicaStatusTxt.contains("Primary")).isTrue(); + assertThat(replicaStatusTxt.contains("P")).isTrue(); assertThat(replicaStatusTxt.contains("Connected")).isTrue(); // validate serviceEndpointStatistics JsonNode serviceEndpointStatistics = storeResult.get("serviceEndpointStatistics"); diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java index e3cde37f96c0..9d0b39e4aa9b 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java @@ -11,7 +11,9 @@ import com.azure.cosmos.implementation.Strings; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.util.function.Tuple2; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; @@ -37,6 +39,28 @@ public Mono> resolveAllUriAsync( .map(a -> a.getPhysicalUri()).collect(Collectors.toList())); } + public Mono>> resolveAndStoreAllUriAsync( + RxDocumentServiceRequest request, + boolean includePrimary, + boolean forceRefresh) { + Mono> allReplicaAddressesObs = this.resolveAddressesAsync(request, forceRefresh); + return allReplicaAddressesObs.map(allReplicaAddresses -> { + List readReplicas = new ArrayList<>(); + List allReplicas = new ArrayList<>(); + allReplicaAddresses.forEach(a -> { + if (!a.isPrimary() || includePrimary) { + readReplicas.add(a.getPhysicalUri()); + } + allReplicas.add(a.getPhysicalUri()); + }); + List> replicasTuple = new ArrayList<>(); + replicasTuple.add(readReplicas); + replicasTuple.add(allReplicas); + + return replicasTuple; + }); + } + public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolean forceAddressRefresh) { Mono> replicaAddressesObs = this.resolveAddressesAsync(request, forceAddressRefresh); return replicaAddressesObs.flatMap(replicaAddresses -> { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java index 8ba1ab0f94ac..105b97f35db1 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ConsistencyWriter.java @@ -209,7 +209,7 @@ Mono writePrivateAsync( return Mono.error(e); } replicaStatusList.put(Uri.IGNORING, replicaStatuses); - replicaStatusList.put(Uri.ATTEMPTING,new HashSet<>(Arrays.asList(primaryUri.getHealthStatusDiagnosticString()))); + replicaStatusList.put(Uri.ATTEMPTING, new HashSet<>(Arrays.asList(primaryUri.getHealthStatusDiagnosticString()))); return this.transportClient.invokeResourceOperationAsync(primaryUri, request) .doOnError( diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java index 542c00b56867..0135fcb26b4a 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java @@ -216,7 +216,8 @@ private Flux> readFromReplicas(List resultCollect final MutableVolatile requestSessionToken, final MutableVolatile hasGoneException, boolean enforceSessionCheck, - final MutableVolatile shortCircut) { + final MutableVolatile shortCircut, + List allApiResults) { if (entity.requestContext.timeoutHelper.isElapsed()) { return Flux.error(new GoneException()); } @@ -231,7 +232,7 @@ private Flux> readFromReplicas(List resultCollect replicaStatusesAttempting.add(uri.getHealthStatusDiagnosticString()); }); Map> replicaStatusList = new HashMap<>(); - resolveApiResults.stream().filter(uri -> !replicaStatusesAttempting.contains(uri.getHealthStatusDiagnosticString())).forEach(uri -> + allApiResults.stream().filter(uri -> !replicaStatusesAttempting.contains(uri.getHealthStatusDiagnosticString())).forEach(uri -> replicaStatusesNotAttempting.add(uri.getHealthStatusDiagnosticString()) ); replicaStatusList.put(Uri.ATTEMPTING , replicaStatusesAttempting); @@ -417,7 +418,7 @@ private Mono readMultipleReplicasInternalAsync(RxDocumentServ requestedCollectionId = entity.requestContext.resolvedCollectionRid; } - Mono> resolveApiResultsObs = this.addressSelector.resolveAllUriAsync( + Mono>> resolveApiResultsObs = this.addressSelector.resolveAndStoreAllUriAsync( entity, includePrimary, entity.requestContext.forceRefreshAddressCache); @@ -431,8 +432,10 @@ private Mono readMultipleReplicasInternalAsync(RxDocumentServ return resolveApiResultsObs.flux() .map(list -> Collections.synchronizedList(new ArrayList<>(list))) .flatMap( - resolveApiResults -> { + resolveApiResultsTuple -> { try { + List resolveApiResults = resolveApiResultsTuple.get(0); + List allApiResults = resolveApiResultsTuple.get(1); MutableVolatile requestSessionToken = new MutableVolatile<>(); if (useSessionToken) { SessionTokenHelper.setPartitionLocalSessionToken(entity, this.sessionContainer); @@ -476,7 +479,8 @@ private Mono readMultipleReplicasInternalAsync(RxDocumentServ requestSessionToken, hasGoneException, enforceSessionCheck, - shortCircuitResult)) + shortCircuitResult, + allApiResults)) // repeat().takeUntil() simulate a while loop pattern .repeat() .takeUntil(x -> { @@ -775,7 +779,7 @@ StoreResult createAndRecordStoreResult( logger.error("Unexpected failure while recording response", e); } - if (responseException !=null) { + if (responseException != null) { verifyCanContinueOnException(storeResult.getException()); } 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 4b1872c39b41..98919785c940 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 @@ -15,12 +15,10 @@ import org.slf4j.LoggerFactory; import java.io.IOException; -import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkArgument; diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java index bbc605eaa610..7956efcb8d0c 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/Uri.java @@ -61,10 +61,8 @@ public String getURIAsString() { public void setPrimary(boolean primary) { isPrimary = primary; if (primary) { - this.healthStatusTuple.updateAndGet(previousStatusTuple -> { - previousStatusTuple.diagnosticString = previousStatusTuple.diagnosticString.replace(SECONDARY, PRIMARY); - return previousStatusTuple; - }); + this.healthStatusTuple.updateAndGet(previousStatusTuple -> + new HealthStatusAndDiagnosticStringTuple(this.uri, previousStatusTuple.status, true)); } } From fcdbbb84c8c5044c7fbffe9a541b3155c2542849 Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Thu, 13 Jun 2024 13:00:21 -0700 Subject: [PATCH 11/16] addressed comment and updated changelog --- .../AddressSelectorWrapper.java | 53 ++++++++++++++++++- sdk/cosmos/azure-cosmos/CHANGELOG.md | 2 +- .../directconnectivity/AddressSelector.java | 30 +++-------- .../directconnectivity/StoreReader.java | 11 ++-- 4 files changed, 67 insertions(+), 29 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java index 0c38b904dab4..813c2765b1d1 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/directconnectivity/AddressSelectorWrapper.java @@ -15,6 +15,7 @@ import reactor.core.publisher.Mono; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; @@ -268,7 +269,7 @@ public AddressSelectorWrapper verifyResolveAddressesAsync(int count) { } public AddressSelectorWrapper verifyResolveAllUriAsync(int count) { - Mockito.verify(addressSelector, Mockito.times(count)).resolveAllUriAsync(Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean()); + Mockito.verify(addressSelector, Mockito.times(count)).resolveAllUriAsync(Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyList()); return this; } @@ -421,6 +422,41 @@ public AddressSelectorWrapper build() { })).when(addressSelector).resolveAllUriAsync(Mockito.any(RxDocumentServiceRequest.class), Mockito.anyBoolean(), Mockito.anyBoolean()); + Mockito.doAnswer((invocation -> { + capture(invocation); + RxDocumentServiceRequest request = invocation.getArgument(0, RxDocumentServiceRequest.class); + boolean includePrimary = invocation.getArgument(1, Boolean.class); + boolean forceRefresh = invocation.getArgument(2, Boolean.class); + List allReplicas = invocation.getArgument(3, List.class); + ImmutableList.Builder b = ImmutableList.builder(); + + if (forceRefresh || refreshed.get()) { + if (partitionKeyRangeFunction != null) { + request.requestContext.resolvedPartitionKeyRange = partitionKeyRangeFunction.apply(request); + } + refreshed.set(true); + if (includePrimary) { + b.add(primary.getRight()); + } + allReplicas = ImmutableList.builder().addAll(secondary.stream().map(s -> s.getRight()).collect(Collectors.toList())) + .add(primary.getRight()).build(); + b.addAll(secondary.stream().map(s -> s.getRight()).collect(Collectors.toList())); + return Mono.just(b.build()); + } else { + // old + if (includePrimary) { + b.add(primary.getLeft()); + } + + allReplicas = ImmutableList.builder().addAll(secondary.stream().map(s -> s.getLeft()).collect(Collectors.toList())) + .add(primary.getLeft()).build(); + b.addAll(secondary.stream().map(s -> s.getLeft()).collect(Collectors.toList())); + return Mono.just(b.build()); + } + + })).when(addressSelector).resolveAllUriAsync(Mockito.any(RxDocumentServiceRequest.class), Mockito.anyBoolean(), + Mockito.anyBoolean(), Mockito.anyList()); + Mockito.doAnswer((invocation -> { capture(invocation); RxDocumentServiceRequest request = invocation.getArgument(0, RxDocumentServiceRequest.class); @@ -497,6 +533,21 @@ public AddressSelectorWrapper build() { } })).when(addressSelector).resolveAllUriAsync(Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean()); + Mockito.doAnswer((invocation -> { + capture(invocation); + RxDocumentServiceRequest request = invocation.getArgument(0, RxDocumentServiceRequest.class); + boolean includePrimary = invocation.getArgument(1, Boolean.class); + boolean forceRefresh = invocation.getArgument(2, Boolean.class); + List allReplicas = invocation.getArgument(3, List.class); + + allReplicas = ImmutableList.builder().addAll(secondaryAddresses).add(primaryAddress).build(); + if (includePrimary) { + return Mono.just(ImmutableList.builder().addAll(secondaryAddresses).add(primaryAddress).build()); + } else { + return Mono.just(secondaryAddresses); + } + })).when(addressSelector).resolveAllUriAsync(Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyList()); + Mockito.doAnswer((invocation -> { capture(invocation); return Mono.just(ImmutableList.builder() diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index 9f91d47a5f59..9977363f8543 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -9,6 +9,7 @@ #### Bugs Fixed #### Other Changes +* Added diagnostic fields for `quorumAckedLSN` and `currentReplicaSetSize`. Changed `replicaStatusList` to include all replicas and more information. - See [PR 39844](https://github.com/Azure/azure-sdk-for-java/pull/39844) ### 4.61.1 (2024-05-31) @@ -46,7 +47,6 @@ #### Other Changes * Load Blackbird or Afterburner into the ObjectMapper depending upon Java version and presence of modules in classpath. Make Afterburner and Blackbird optional maven dependencies. See - [PR 39689](https://github.com/Azure/azure-sdk-for-java/pull/39689) -* Added diagnostic fields for `quorumAckedLSN` and `currentReplicaSetSize`. Changed `replicaStatusList` to include all replicas and more information. - See [PR 39844](https://github.com/Azure/azure-sdk-for-java/pull/39844) ### 4.53.5-hotfix (2024-04-25) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java index 9d0b39e4aa9b..a62388c891ec 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/AddressSelector.java @@ -4,21 +4,16 @@ package com.azure.cosmos.implementation.directconnectivity; import com.azure.cosmos.CosmosContainerProactiveInitConfig; -import com.azure.cosmos.implementation.AsyncDocumentClient; import com.azure.cosmos.implementation.GoneException; -import com.azure.cosmos.implementation.HttpConstants; import com.azure.cosmos.implementation.RxDocumentServiceRequest; import com.azure.cosmos.implementation.Strings; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.util.function.Tuple2; import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import java.util.Map; import java.util.Set; -import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; public class AddressSelector { @@ -39,26 +34,17 @@ public Mono> resolveAllUriAsync( .map(a -> a.getPhysicalUri()).collect(Collectors.toList())); } - public Mono>> resolveAndStoreAllUriAsync( + public Mono> resolveAllUriAsync( RxDocumentServiceRequest request, boolean includePrimary, - boolean forceRefresh) { + boolean forceRefresh, + List allReplicas) { Mono> allReplicaAddressesObs = this.resolveAddressesAsync(request, forceRefresh); - return allReplicaAddressesObs.map(allReplicaAddresses -> { - List readReplicas = new ArrayList<>(); - List allReplicas = new ArrayList<>(); - allReplicaAddresses.forEach(a -> { - if (!a.isPrimary() || includePrimary) { - readReplicas.add(a.getPhysicalUri()); - } - allReplicas.add(a.getPhysicalUri()); - }); - List> replicasTuple = new ArrayList<>(); - replicasTuple.add(readReplicas); - replicasTuple.add(allReplicas); - - return replicasTuple; - }); + return allReplicaAddressesObs.map(allReplicaAddresses -> allReplicaAddresses.stream().map(a -> { + allReplicas.add(a.getPhysicalUri()); + return a; + }).filter(a -> includePrimary || !a.isPrimary()) + .map(a -> a.getPhysicalUri()).collect(Collectors.toList())); } public Mono resolvePrimaryUriAsync(RxDocumentServiceRequest request, boolean forceAddressRefresh) { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java index 0135fcb26b4a..459b3979c5f4 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreReader.java @@ -418,10 +418,13 @@ private Mono readMultipleReplicasInternalAsync(RxDocumentServ requestedCollectionId = entity.requestContext.resolvedCollectionRid; } - Mono>> resolveApiResultsObs = this.addressSelector.resolveAndStoreAllUriAsync( + List allApiResults = new ArrayList<>(); + + Mono> resolveApiResultsObs = this.addressSelector.resolveAllUriAsync( entity, includePrimary, - entity.requestContext.forceRefreshAddressCache); + entity.requestContext.forceRefreshAddressCache, + allApiResults); if (!StringUtils.isEmpty(requestedCollectionId) && !StringUtils.isEmpty(entity.requestContext.resolvedCollectionRid)) { if (!requestedCollectionId.equals(entity.requestContext.resolvedCollectionRid)) { @@ -432,10 +435,8 @@ private Mono readMultipleReplicasInternalAsync(RxDocumentServ return resolveApiResultsObs.flux() .map(list -> Collections.synchronizedList(new ArrayList<>(list))) .flatMap( - resolveApiResultsTuple -> { + resolveApiResults -> { try { - List resolveApiResults = resolveApiResultsTuple.get(0); - List allApiResults = resolveApiResultsTuple.get(1); MutableVolatile requestSessionToken = new MutableVolatile<>(); if (useSessionToken) { SessionTokenHelper.setPartitionLocalSessionToken(entity, this.sessionContainer); From 54c3a40c553a65768461608bf42dee63e106cc09 Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Mon, 8 Jul 2024 10:06:48 -0700 Subject: [PATCH 12/16] fixed regions contacted issue --- .../java/com/azure/cosmos/CosmosDiagnosticsTest.java | 12 ++++++------ .../cosmos/implementation/routing/LocationCache.java | 2 +- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java index f692a2433031..87707cf4e63a 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java @@ -1725,15 +1725,15 @@ private void validateRegionContacted(CosmosDiagnostics cosmosDiagnostics, Cosmos Class DatabaseAccountLocationsInfoClass = Class.forName("com.azure.cosmos.implementation.routing" + ".LocationCache$DatabaseAccountLocationsInfo"); - Field availableWriteEndpointByLocation = DatabaseAccountLocationsInfoClass.getDeclaredField( - "availableWriteEndpointByLocation"); - availableWriteEndpointByLocation.setAccessible(true); + Field availableWriteLocations = DatabaseAccountLocationsInfoClass.getDeclaredField( + "availableWriteLocations"); + availableWriteLocations.setAccessible(true); @SuppressWarnings("unchecked") - Map map = (Map) availableWriteEndpointByLocation.get(locationInfo); - String regionName = map.keySet().iterator().next(); + List list = (List) availableWriteLocations.get(locationInfo); + String regionName = list.get(0); assertThat(cosmosDiagnostics.getContactedRegionNames().size()).isEqualTo(1); - assertThat(cosmosDiagnostics.getContactedRegionNames().iterator().next()).isEqualTo(regionName.toLowerCase()); + assertThat(cosmosDiagnostics.getContactedRegionNames().iterator().next()).isEqualTo(regionName); } private void validateChannelStatistics(CosmosDiagnostics cosmosDiagnostics) { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/routing/LocationCache.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/routing/LocationCache.java index a7bbbfc70855..1f33840da9a7 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/routing/LocationCache.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/routing/LocationCache.java @@ -422,7 +422,7 @@ public String getRegionName(URI locationEndpoint, com.azure.cosmos.implementatio } //If preferred list is not set, locationEndpoint will be default endpoint, so return the hub region - return this.locationInfo.availableWriteEndpointByLocation.keySet().iterator().next(); + return this.locationInfo.availableWriteLocations.get(0); } private boolean areEqual(URI url1, URI url2) { From a77598c02b4f09c3452c80588be119f84782e8f2 Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Mon, 8 Jul 2024 10:15:32 -0700 Subject: [PATCH 13/16] updated changelog --- sdk/cosmos/azure-cosmos/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index 1586cdb905e0..9dee72c01e02 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -7,9 +7,9 @@ #### Breaking Changes #### Bugs Fixed +* Fixed an issue where `contactedRegions` shows the wrong region in a multi region account if no preferred regions are specified. - See [PR 41107]() #### Other Changes -* Added diagnostic fields for `quorumAckedLSN` and `currentReplicaSetSize`. Changed `replicaStatusList` to include all replicas and more information. - See [PR 39844](https://github.com/Azure/azure-sdk-for-java/pull/39844) ### 4.62.0 (2024-07-02) From 758e0cafff59d65e9a60418a18e277d93e8e2dd8 Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Mon, 8 Jul 2024 10:37:43 -0700 Subject: [PATCH 14/16] Update sdk/cosmos/azure-cosmos/CHANGELOG.md --- sdk/cosmos/azure-cosmos/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index 9dee72c01e02..f6b575f4696e 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -7,7 +7,7 @@ #### Breaking Changes #### Bugs Fixed -* Fixed an issue where `contactedRegions` shows the wrong region in a multi region account if no preferred regions are specified. - See [PR 41107]() +* Fixed an issue where `contactedRegions` shows the wrong region in a multi region account if no preferred regions are specified. - See [PR 41045](https://github.com/Azure/azure-sdk-for-java/pull/41045/files) #### Other Changes From 0407a8c5d366517d7764b6d95cd8f35258549332 Mon Sep 17 00:00:00 2001 From: Fabian Meiswinkel Date: Mon, 8 Jul 2024 20:15:07 +0200 Subject: [PATCH 15/16] Update sdk/cosmos/azure-cosmos/CHANGELOG.md Co-authored-by: Kushagra Thapar --- sdk/cosmos/azure-cosmos/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index f6b575f4696e..e91396739bb0 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -7,7 +7,7 @@ #### Breaking Changes #### Bugs Fixed -* Fixed an issue where `contactedRegions` shows the wrong region in a multi region account if no preferred regions are specified. - See [PR 41045](https://github.com/Azure/azure-sdk-for-java/pull/41045/files) +* Fixed an issue where `contactedRegions` shows the wrong region in a multi region account if no preferred regions are specified. - See [PR 41045](https://github.com/Azure/azure-sdk-for-java/pull/41045) #### Other Changes From 76a0b9f4d828fb20026da02f770843daaf026fdf Mon Sep 17 00:00:00 2001 From: Tomas Varon <70857381+tvaron3@users.noreply.github.com> Date: Wed, 10 Jul 2024 10:27:54 -0700 Subject: [PATCH 16/16] stabilizing live tests --- .../com/azure/cosmos/ClientMetricsTest.java | 33 +++++++------------ .../azure/cosmos/CosmosDiagnosticsTest.java | 2 +- .../SessionNotAvailableRetryTest.java | 27 +++++---------- .../implementation/routing/LocationCache.java | 2 +- 4 files changed, 21 insertions(+), 43 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/ClientMetricsTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/ClientMetricsTest.java index 27a1e4456001..19b2bbc3cc30 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/ClientMetricsTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/ClientMetricsTest.java @@ -68,8 +68,6 @@ import java.util.EnumSet; import java.util.List; import java.util.Locale; -import java.util.Map; -import java.util.Set; import java.util.UUID; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; @@ -146,7 +144,7 @@ public void beforeTest( AsyncDocumentClient asyncDocumentClient = ReflectionUtils.getAsyncDocumentClient(this.client.asyncClient()); RxDocumentClientImpl rxDocumentClient = (RxDocumentClientImpl) asyncDocumentClient; - Set writeRegions = this.getAvailableRegionNames(rxDocumentClient, true); + List writeRegions = this.getAvailableWriteRegionNames(rxDocumentClient); assertThat(writeRegions).isNotNull().isNotEmpty(); this.preferredRegion = writeRegions.iterator().next(); @@ -1511,7 +1509,7 @@ private void validateMetrics(Tag expectedOperationTag, Tag expectedRequestTag, i this.assertMetrics( "cosmos.client.op.regionsContacted", true, - Tag.of(TagName.RegionName.toString(), this.preferredRegion)); + Tag.of(TagName.RegionName.toString(), this.preferredRegion.toLowerCase(Locale.ROOT))); } if (this.getEffectiveMetricCategories().contains(MetricCategory.RequestSummary)) { @@ -1531,7 +1529,7 @@ private void validateMetrics(Tag expectedOperationTag, Tag expectedRequestTag, i this.assertMetrics( "cosmos.client.req.rntbd.latency", true, - Tag.of(TagName.RegionName.toString(), this.preferredRegion)); + Tag.of(TagName.RegionName.toString(), this.preferredRegion.toLowerCase(Locale.ROOT))); this.assertMetrics("cosmos.client.req.rntbd.backendLatency", true, expectedRequestTag); this.assertMetrics("cosmos.client.req.rntbd.requests", true, expectedRequestTag); Meter reportedRntbdRequestCharge = @@ -1548,7 +1546,7 @@ private void validateMetrics(Tag expectedOperationTag, Tag expectedRequestTag, i this.assertMetrics( "cosmos.client.req.gw.latency", true, - Tag.of(TagName.RegionName.toString(), this.preferredRegion)); + Tag.of(TagName.RegionName.toString(), this.preferredRegion.toLowerCase(Locale.ROOT))); } this.assertMetrics("cosmos.client.req.gw.backendLatency", false, expectedRequestTag); this.assertMetrics("cosmos.client.req.gw.requests", true, expectedRequestTag); @@ -1654,7 +1652,7 @@ private Meter assertMetrics(String prefix, boolean expectedToFind, Tag withTag) } } - private Set getAvailableRegionNames(RxDocumentClientImpl rxDocumentClient, boolean isWriteRegion) { + private List getAvailableWriteRegionNames(RxDocumentClientImpl rxDocumentClient) { try { GlobalEndpointManager globalEndpointManager = ReflectionUtils.getGlobalEndpointManager(rxDocumentClient); LocationCache locationCache = ReflectionUtils.getLocationCache(globalEndpointManager); @@ -1665,22 +1663,13 @@ private Set getAvailableRegionNames(RxDocumentClientImpl rxDocumentClien Class DatabaseAccountLocationsInfoClass = Class.forName("com.azure.cosmos.implementation.routing" + ".LocationCache$DatabaseAccountLocationsInfo"); + Field availableWriteLocations = DatabaseAccountLocationsInfoClass.getDeclaredField( + "availableWriteLocations"); + availableWriteLocations.setAccessible(true); + @SuppressWarnings("unchecked") + List list = (List) availableWriteLocations.get(locationInfo); + return list; - if (isWriteRegion) { - Field availableWriteEndpointByLocation = DatabaseAccountLocationsInfoClass.getDeclaredField( - "availableWriteEndpointByLocation"); - availableWriteEndpointByLocation.setAccessible(true); - @SuppressWarnings("unchecked") - Map map = (Map) availableWriteEndpointByLocation.get(locationInfo); - return map.keySet(); - } else { - Field availableReadEndpointByLocation = DatabaseAccountLocationsInfoClass.getDeclaredField( - "availableReadEndpointByLocation"); - availableReadEndpointByLocation.setAccessible(true); - @SuppressWarnings("unchecked") - Map map = (Map) availableReadEndpointByLocation.get(locationInfo); - return map.keySet(); - } } catch (Exception error) { fail(error.toString()); diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java index 87707cf4e63a..6e1d15752f82 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosDiagnosticsTest.java @@ -1733,7 +1733,7 @@ private void validateRegionContacted(CosmosDiagnostics cosmosDiagnostics, Cosmos List list = (List) availableWriteLocations.get(locationInfo); String regionName = list.get(0); assertThat(cosmosDiagnostics.getContactedRegionNames().size()).isEqualTo(1); - assertThat(cosmosDiagnostics.getContactedRegionNames().iterator().next()).isEqualTo(regionName); + assertThat(cosmosDiagnostics.getContactedRegionNames().iterator().next()).isEqualTo(regionName.toLowerCase()); } private void validateChannelStatistics(CosmosDiagnostics cosmosDiagnostics) { diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/SessionNotAvailableRetryTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/SessionNotAvailableRetryTest.java index 7f19b29db84f..69b2f05aa862 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/SessionNotAvailableRetryTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/SessionNotAvailableRetryTest.java @@ -43,7 +43,6 @@ import reactor.core.publisher.Mono; import java.lang.reflect.Field; -import java.net.URI; import java.util.ArrayList; import java.util.Arrays; import java.util.HashSet; @@ -51,7 +50,6 @@ import java.util.List; import java.util.Locale; import java.util.Map; -import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; @@ -354,7 +352,7 @@ public void sessionNotAvailableRetryWithoutPreferredList(OperationType operation } catch (CosmosException ex) { assertThat(ex.getStatusCode()).isEqualTo(HttpConstants.StatusCodes.NOTFOUND); assertThat(ex.getDiagnostics().getContactedRegionNames().size()).isEqualTo(1); - String regionName = getAvailableRegionNames(rxDocumentClient, true).iterator().next(); + String regionName = getRegionNames(rxDocumentClient).iterator().next(); assertThat(ex.getDiagnostics().getContactedRegionNames().iterator().next()).isEqualTo(regionName.toLowerCase()); } @@ -409,7 +407,7 @@ private Map getRegionMap(DatabaseAccount databaseAccount, boolea return regionMap; } - private Set getAvailableRegionNames(RxDocumentClientImpl rxDocumentClient, boolean isWriteRegion) throws Exception { + private List getRegionNames(RxDocumentClientImpl rxDocumentClient) throws Exception { GlobalEndpointManager globalEndpointManager = ReflectionUtils.getGlobalEndpointManager(rxDocumentClient); LocationCache locationCache = ReflectionUtils.getLocationCache(globalEndpointManager); @@ -420,21 +418,12 @@ private Set getAvailableRegionNames(RxDocumentClientImpl rxDocumentClien Class DatabaseAccountLocationsInfoClass = Class.forName("com.azure.cosmos.implementation.routing" + ".LocationCache$DatabaseAccountLocationsInfo"); - if (isWriteRegion) { - Field availableWriteEndpointByLocation = DatabaseAccountLocationsInfoClass.getDeclaredField( - "availableWriteEndpointByLocation"); - availableWriteEndpointByLocation.setAccessible(true); - @SuppressWarnings("unchecked") - Map map = (Map) availableWriteEndpointByLocation.get(locationInfo); - return map.keySet(); - } else { - Field availableReadEndpointByLocation = DatabaseAccountLocationsInfoClass.getDeclaredField( - "availableReadEndpointByLocation"); - availableReadEndpointByLocation.setAccessible(true); - @SuppressWarnings("unchecked") - Map map = (Map) availableReadEndpointByLocation.get(locationInfo); - return map.keySet(); - } + Field availableWriteEndpointByLocation = DatabaseAccountLocationsInfoClass.getDeclaredField( + "availableWriteLocations"); + availableWriteEndpointByLocation.setAccessible(true); + @SuppressWarnings("unchecked") + List list = (List) availableWriteEndpointByLocation.get(locationInfo); + return list; } private class RntbdTransportClientTest extends TransportClient { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/routing/LocationCache.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/routing/LocationCache.java index 1f33840da9a7..30ad1696aa5e 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/routing/LocationCache.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/routing/LocationCache.java @@ -422,7 +422,7 @@ public String getRegionName(URI locationEndpoint, com.azure.cosmos.implementatio } //If preferred list is not set, locationEndpoint will be default endpoint, so return the hub region - return this.locationInfo.availableWriteLocations.get(0); + return this.locationInfo.availableWriteLocations.get(0).toLowerCase(Locale.ROOT); } private boolean areEqual(URI url1, URI url2) {