diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/GlobalEndpointManager.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/GlobalEndpointManager.java index dcc3762eaf3d..2bd3e9a6c0d9 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/GlobalEndpointManager.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/GlobalEndpointManager.java @@ -3,12 +3,14 @@ package com.azure.cosmos.implementation; -import com.azure.cosmos.implementation.routing.LocationCache; -import com.azure.cosmos.implementation.routing.LocationHelper; import com.azure.cosmos.BridgeInternal; import com.azure.cosmos.ConnectionPolicy; import com.azure.cosmos.DatabaseAccount; +import com.azure.cosmos.implementation.caches.AsyncCache; +import com.azure.cosmos.implementation.routing.LocationCache; +import com.azure.cosmos.implementation.routing.LocationHelper; import org.apache.commons.collections4.list.UnmodifiableList; +import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import reactor.core.publisher.Flux; @@ -16,13 +18,13 @@ import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; -import java.net.URISyntaxException; import java.net.URI; import java.time.Duration; import java.time.LocalDateTime; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean; @@ -45,6 +47,8 @@ public class GlobalEndpointManager implements AutoCloseable { private final ExecutorService executor = Executors.newSingleThreadExecutor(); private final Scheduler scheduler = Schedulers.fromExecutor(executor); private volatile boolean isClosed; + private AtomicBoolean firstTimeDatabaseAccountInitialization = new AtomicBoolean(true); + private volatile DatabaseAccount latestDatabaseAccount; public GlobalEndpointManager(DatabaseAccountManagerInternal owner, ConnectionPolicy connectionPolicy, Configs configs) { this.backgroundRefreshLocationTimeIntervalInMS = configs.getUnavailableLocationsExpirationTimeInSeconds() * 1000; @@ -159,6 +163,16 @@ public Mono refreshLocationAsync(DatabaseAccount databaseAccount, boolean }); } + /** + * This will provide the latest databaseAccount. + * If due to some reason last databaseAccount update was null, + * this method will return previous valid value + * @return DatabaseAccount + */ + public DatabaseAccount getLatestDatabaseAccount() { + return this.latestDatabaseAccount; + } + private Mono refreshLocationPrivateAsync(DatabaseAccount databaseAccount) { return Mono.defer(() -> { logger.debug("refreshLocationPrivateAsync() refreshing locations"); @@ -254,7 +268,13 @@ private Mono startRefreshLocationTimerAsync(boolean initialization) { private Mono getDatabaseAccountAsync(URI serviceEndpoint) { return this.owner.getDatabaseAccountFromEndpoint(serviceEndpoint) - .doOnNext(i -> logger.debug("account retrieved: {}", i)).single(); + .doOnNext(databaseAccount -> { + if(databaseAccount != null) { + this.latestDatabaseAccount = databaseAccount; + } + + logger.debug("account retrieved: {}", databaseAccount); + }).single(); } public boolean isClosed() { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java index 1a14d936c996..fbbf744c1ae2 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java @@ -251,29 +251,16 @@ private RxDocumentClientImpl(URI serviceEndpoint, } private void initializeGatewayConfigurationReader() { - String resourceToken; - if(this.tokenResolver != null) { - resourceToken = this.tokenResolver.getAuthorizationToken(RequestVerb.GET, "", CosmosResourceType.System, null); - } else if(!this.hasAuthKeyResourceToken && this.authorizationTokenProvider == null) { - resourceToken = this.firstResourceTokenFromPermissionFeed; - } else { - assert this.masterKeyOrResourceToken != null || this.cosmosKeyCredential != null; - resourceToken = this.masterKeyOrResourceToken; - } - - this.gatewayConfigurationReader = new GatewayServiceConfigurationReader(this.serviceEndpoint, - this.hasAuthKeyResourceToken, - resourceToken, - this.connectionPolicy, - this.authorizationTokenProvider, - this.reactorHttpClient); - - DatabaseAccount databaseAccount = this.gatewayConfigurationReader.initializeReaderAsync().block(); + this.gatewayConfigurationReader = new GatewayServiceConfigurationReader(this.globalEndpointManager); + DatabaseAccount databaseAccount = this.globalEndpointManager.getLatestDatabaseAccount(); + //Database account should not be null here, + // this.globalEndpointManager.init() must have been already called + // hence asserting it + assert(databaseAccount != null); this.useMultipleWriteLocations = this.connectionPolicy.getUsingMultipleWriteLocations() && BridgeInternal.isEnableMultipleWriteLocations(databaseAccount); // TODO: add support for openAsync // https://msdata.visualstudio.com/CosmosDB/_workitems/edit/332589 - this.globalEndpointManager.refreshLocationAsync(databaseAccount, false).block(); } public void init() { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfigurationReader.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfigurationReader.java index d9f562885cac..adb678c0788b 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfigurationReader.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfigurationReader.java @@ -4,29 +4,11 @@ package com.azure.cosmos.implementation.directconnectivity; import com.azure.cosmos.BridgeInternal; -import com.azure.cosmos.ConnectionPolicy; import com.azure.cosmos.ConsistencyLevel; -import com.azure.cosmos.RequestVerb; -import com.azure.cosmos.implementation.BaseAuthorizationTokenProvider; -import com.azure.cosmos.implementation.Constants; -import com.azure.cosmos.DatabaseAccount; import com.azure.cosmos.implementation.GlobalEndpointManager; -import com.azure.cosmos.implementation.HttpConstants; import com.azure.cosmos.implementation.ReplicationPolicy; -import com.azure.cosmos.implementation.UserAgentContainer; -import com.azure.cosmos.implementation.Utils; -import com.azure.cosmos.implementation.http.HttpClient; -import com.azure.cosmos.implementation.http.HttpHeaders; -import com.azure.cosmos.implementation.http.HttpRequest; -import com.azure.cosmos.implementation.http.HttpResponse; -import io.netty.handler.codec.http.HttpMethod; -import reactor.core.publisher.Mono; -import java.net.MalformedURLException; import java.net.URI; -import java.net.URISyntaxException; -import java.util.ArrayList; -import java.util.HashMap; import java.util.Map; /** @@ -42,118 +24,25 @@ */ public class GatewayServiceConfigurationReader { - public static final String GATEWAY_READER_NOT_INITIALIZED = "GatewayServiceConfigurationReader has not been initialized"; + private GlobalEndpointManager globalEndpointManager; - public ReplicationPolicy userReplicationPolicy; - private ReplicationPolicy systemReplicationPolicy; - private ConsistencyLevel consistencyLevel; - private volatile boolean initialized; - private URI serviceEndpoint; - private final ConnectionPolicy connectionPolicy; - private Map queryEngineConfiguration; - private final BaseAuthorizationTokenProvider baseAuthorizationTokenProvider; - private final boolean hasAuthKeyResourceToken; - private final String authKeyResourceToken; - private HttpClient httpClient; - - public GatewayServiceConfigurationReader(URI serviceEndpoint, boolean hasResourceToken, String resourceToken, - ConnectionPolicy connectionPolicy, BaseAuthorizationTokenProvider baseAuthorizationTokenProvider, - HttpClient httpClient) { - this.serviceEndpoint = serviceEndpoint; - this.baseAuthorizationTokenProvider = baseAuthorizationTokenProvider; - this.hasAuthKeyResourceToken = hasResourceToken; - this.authKeyResourceToken = resourceToken; - this.connectionPolicy = connectionPolicy; - this.httpClient = httpClient; + public GatewayServiceConfigurationReader(GlobalEndpointManager globalEndpointManager) { + this.globalEndpointManager = globalEndpointManager; } public ReplicationPolicy getUserReplicationPolicy() { - this.throwIfNotInitialized(); - return this.userReplicationPolicy; + return BridgeInternal.getReplicationPolicy(this.globalEndpointManager.getLatestDatabaseAccount()); } public ReplicationPolicy getSystemReplicationPolicy() { - this.throwIfNotInitialized(); - return this.systemReplicationPolicy; - } - - public boolean enableAuthorization() { - return true; + return BridgeInternal.getSystemReplicationPolicy(this.globalEndpointManager.getLatestDatabaseAccount()); } public ConsistencyLevel getDefaultConsistencyLevel() { - this.throwIfNotInitialized(); - return this.consistencyLevel; - } - - public void setDefaultConsistencyLevel(ConsistencyLevel value) { - this.throwIfNotInitialized(); - this.consistencyLevel = value; + return BridgeInternal.getConsistencyPolicy(this.globalEndpointManager.getLatestDatabaseAccount()).getDefaultConsistencyLevel(); } public Map getQueryEngineConfiguration() { - this.throwIfNotInitialized(); - return this.queryEngineConfiguration; - } - - private Mono getDatabaseAccountAsync(URI serviceEndpoint) { - - HttpHeaders httpHeaders = new HttpHeaders(); - httpHeaders.set(HttpConstants.HttpHeaders.VERSION, HttpConstants.Versions.CURRENT_VERSION); - - UserAgentContainer userAgentContainer = new UserAgentContainer(); - String userAgentSuffix = this.connectionPolicy.getUserAgentSuffix(); - if (userAgentSuffix != null && userAgentSuffix.length() > 0) { - userAgentContainer.setSuffix(userAgentSuffix); - } - - httpHeaders.set(HttpConstants.HttpHeaders.USER_AGENT, userAgentContainer.getUserAgent()); - httpHeaders.set(HttpConstants.HttpHeaders.API_TYPE, Constants.Properties.SQL_API_TYPE); - - String xDate = Utils.nowAsRFC1123(); - httpHeaders.set(HttpConstants.HttpHeaders.X_DATE, xDate); - - String authorizationToken; - if (this.hasAuthKeyResourceToken || baseAuthorizationTokenProvider == null) { - authorizationToken = HttpUtils.urlEncode(this.authKeyResourceToken); - } else { - // Retrieve the document service properties. - Map header = new HashMap<>(); - header.put(HttpConstants.HttpHeaders.X_DATE, xDate); - authorizationToken = baseAuthorizationTokenProvider - .generateKeyAuthorizationSignature(RequestVerb.GET, serviceEndpoint, header); - } - httpHeaders.set(HttpConstants.HttpHeaders.AUTHORIZATION, authorizationToken); - - HttpRequest httpRequest = new HttpRequest(HttpMethod.GET, serviceEndpoint, serviceEndpoint.getPort(), httpHeaders); - Mono httpResponse = httpClient.send(httpRequest); - return toDatabaseAccountObservable(httpResponse, httpRequest); - } - - public Mono initializeReaderAsync() { - return GlobalEndpointManager.getDatabaseAccountFromAnyLocationsAsync(this.serviceEndpoint, - - new ArrayList<>(this.connectionPolicy.getPreferredLocations()), url -> { - return getDatabaseAccountAsync(url); - - }).doOnSuccess(databaseAccount -> { - userReplicationPolicy = BridgeInternal.getReplicationPolicy(databaseAccount); - systemReplicationPolicy = BridgeInternal.getSystemReplicationPolicy(databaseAccount); - queryEngineConfiguration = BridgeInternal.getQueryEngineConfiuration(databaseAccount); - consistencyLevel = BridgeInternal.getConsistencyPolicy(databaseAccount).getDefaultConsistencyLevel(); - initialized = true; - }); - } - - private Mono toDatabaseAccountObservable(Mono httpResponse, HttpRequest httpRequest) { - - return HttpClientUtils.parseResponseAsync(httpResponse, httpRequest) - .map(rxDocumentServiceResponse -> rxDocumentServiceResponse.getResource(DatabaseAccount.class)); - } - - private void throwIfNotInitialized() { - if (!this.initialized) { - throw new IllegalArgumentException(GATEWAY_READER_NOT_INITIALIZED); - } + return BridgeInternal.getQueryEngineConfiuration(this.globalEndpointManager.getLatestDatabaseAccount()); } } diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfigurationReaderTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfigurationReaderTest.java index d6e1d9f2f73e..48b1a45be73c 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfigurationReaderTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfigurationReaderTest.java @@ -5,33 +5,28 @@ import com.azure.cosmos.BridgeInternal; import com.azure.cosmos.ConnectionPolicy; -import com.azure.cosmos.CosmosKeyCredential; +import com.azure.cosmos.ConsistencyLevel; import com.azure.cosmos.DatabaseAccount; import com.azure.cosmos.implementation.AsyncDocumentClient; import com.azure.cosmos.implementation.AsyncDocumentClient.Builder; import com.azure.cosmos.implementation.BaseAuthorizationTokenProvider; -import com.azure.cosmos.implementation.SpyClientUnderTestFactory; +import com.azure.cosmos.implementation.Configs; +import com.azure.cosmos.implementation.DatabaseAccountManagerInternal; +import com.azure.cosmos.implementation.GlobalEndpointManager; +import com.azure.cosmos.implementation.HttpConstants; +import com.azure.cosmos.implementation.RxDocumentClientImpl; import com.azure.cosmos.implementation.TestConfigurations; import com.azure.cosmos.implementation.TestSuiteBase; import com.azure.cosmos.implementation.Utils; import com.azure.cosmos.implementation.http.HttpClient; -import com.azure.cosmos.implementation.http.HttpHeaders; -import com.azure.cosmos.implementation.http.HttpRequest; -import com.azure.cosmos.implementation.http.HttpResponse; -import io.netty.buffer.ByteBufAllocator; -import io.netty.buffer.ByteBufUtil; -import io.reactivex.subscribers.TestSubscriber; -import org.apache.commons.io.IOUtils; +import org.mockito.Matchers; import org.mockito.Mockito; import org.testng.annotations.AfterClass; -import org.testng.annotations.BeforeClass; import org.testng.annotations.Factory; import org.testng.annotations.Test; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; import java.net.URI; -import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; @@ -52,105 +47,57 @@ public GatewayServiceConfigurationReaderTest(Builder clientBuilder) { super(clientBuilder); } - @BeforeClass(groups = "simple") - public void before_GatewayServiceConfigurationReaderTest() throws Exception { - client = clientBuilder().build(); - SpyClientUnderTestFactory.ClientUnderTest clientUnderTest = SpyClientUnderTestFactory.createClientUnderTest(this.clientBuilder()); - HttpClient httpClient = clientUnderTest.getSpyHttpClient(); - baseAuthorizationTokenProvider = new BaseAuthorizationTokenProvider(new CosmosKeyCredential(TestConfigurations.MASTER_KEY)); - connectionPolicy = ConnectionPolicy.getDefaultPolicy(); - mockHttpClient = Mockito.mock(HttpClient.class); - mockGatewayServiceConfigurationReader = new GatewayServiceConfigurationReader(new URI(TestConfigurations.HOST), - false, TestConfigurations.MASTER_KEY, connectionPolicy, baseAuthorizationTokenProvider, mockHttpClient); - - gatewayServiceConfigurationReader = new GatewayServiceConfigurationReader(new URI(TestConfigurations.HOST), - false, - TestConfigurations.MASTER_KEY, - connectionPolicy, - baseAuthorizationTokenProvider, - httpClient); - databaseAccountJson = IOUtils - .toString(getClass().getClassLoader().getResourceAsStream("databaseAccount.json"), "UTF-8"); - expectedDatabaseAccount = new DatabaseAccount(databaseAccountJson); - HttpResponse mockResponse = getMockResponse(databaseAccountJson); - Mockito.when(mockHttpClient.send(Mockito.any(HttpRequest.class))).thenReturn(Mono.just(mockResponse)); - } - @AfterClass(groups = { "simple" }, timeOut = SHUTDOWN_TIMEOUT, alwaysRun = true) public void afterClass() { safeClose(client); } @Test(groups = "simple") - public void mockInitializeReaderAsync() { - Mono databaseAccount = mockGatewayServiceConfigurationReader.initializeReaderAsync(); - validateSuccess(databaseAccount, expectedDatabaseAccount); - } - - @Test(groups = "simple") - public void mockInitializeReaderAsyncWithResourceToken() throws Exception { - HttpResponse mockResponse = getMockResponse(databaseAccountJson); - Mockito.when(mockHttpClient.send(Mockito.any(HttpRequest.class))).thenReturn(Mono.just(mockResponse)); - - mockGatewayServiceConfigurationReader = new GatewayServiceConfigurationReader(new URI(TestConfigurations.HOST), - true, "SampleResourceToken", connectionPolicy, baseAuthorizationTokenProvider, mockHttpClient); - - Mono databaseAccount = mockGatewayServiceConfigurationReader.initializeReaderAsync(); - validateSuccess(databaseAccount, expectedDatabaseAccount); + public void clientInitialization() throws Exception { + client = this.clientBuilder().build(); + RxDocumentClientImpl rxDocumentClient = (RxDocumentClientImpl) client; + GatewayServiceConfigurationReader serviceConfigurationReader = ReflectionUtils.getServiceConfigurationReader(rxDocumentClient); + assertThat(serviceConfigurationReader.getDefaultConsistencyLevel()).isNotNull(); + assertThat(serviceConfigurationReader.getQueryEngineConfiguration()).isNotNull(); + assertThat(serviceConfigurationReader.getSystemReplicationPolicy()).isNotNull(); + assertThat(serviceConfigurationReader.getSystemReplicationPolicy()).isNotNull(); } @Test(groups = "simple") - public void initializeReaderAsync() { - Mono databaseAccount = gatewayServiceConfigurationReader.initializeReaderAsync(); - validateSuccess(databaseAccount); - } - - public static void validateSuccess(Mono observable) { - TestSubscriber testSubscriber = new TestSubscriber(); - - observable.subscribe(testSubscriber); - testSubscriber.awaitTerminalEvent(TIMEOUT, TimeUnit.MILLISECONDS); - testSubscriber.assertNoErrors(); - testSubscriber.assertComplete(); - testSubscriber.assertValueCount(1); - DatabaseAccount databaseAccount = testSubscriber.values().get(0); - assertThat(BridgeInternal.getQueryEngineConfiuration(databaseAccount).size() > 0).isTrue(); - assertThat(BridgeInternal.getReplicationPolicy(databaseAccount)).isNotNull(); - assertThat(BridgeInternal.getSystemReplicationPolicy(databaseAccount)).isNotNull(); - } - - public static void validateSuccess(Mono observable, DatabaseAccount expectedDatabaseAccount) { - TestSubscriber testSubscriber = new TestSubscriber(); - - observable.subscribe(testSubscriber); - testSubscriber.awaitTerminalEvent(TIMEOUT, TimeUnit.MILLISECONDS); - testSubscriber.assertNoErrors(); - testSubscriber.assertComplete(); - testSubscriber.assertValueCount(1); - DatabaseAccount databaseAccount = testSubscriber.values().get(0); - assertThat(databaseAccount.getId()).isEqualTo(expectedDatabaseAccount.getId()); - assertThat(BridgeInternal.getAddressesLink(databaseAccount)) - .isEqualTo(BridgeInternal.getAddressesLink(expectedDatabaseAccount)); - assertThat(databaseAccount.getWritableLocations().iterator().next().getEndpoint()) - .isEqualTo(expectedDatabaseAccount.getWritableLocations().iterator().next().getEndpoint()); - assertThat(BridgeInternal.getSystemReplicationPolicy(databaseAccount).getMaxReplicaSetSize()) - .isEqualTo(BridgeInternal.getSystemReplicationPolicy(expectedDatabaseAccount).getMaxReplicaSetSize()); - assertThat(BridgeInternal.getSystemReplicationPolicy(databaseAccount).getMaxReplicaSetSize()) - .isEqualTo(BridgeInternal.getSystemReplicationPolicy(expectedDatabaseAccount).getMaxReplicaSetSize()); - assertThat(BridgeInternal.getQueryEngineConfiuration(databaseAccount)) - .isEqualTo(BridgeInternal.getQueryEngineConfiuration(expectedDatabaseAccount)); - } - - private HttpResponse getMockResponse(String databaseAccountJson) { - HttpResponse httpResponse = Mockito.mock(HttpResponse.class); - Mockito.doReturn(200).when(httpResponse).statusCode(); - Mockito.doReturn(Flux.just(ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT, databaseAccountJson))) - .when(httpResponse).body(); - Mockito.doReturn(Mono.just(databaseAccountJson)) - .when(httpResponse).bodyAsString(); - Mockito.doReturn(Mono.just(Utils.getUTF8Bytes(databaseAccountJson))) - .when(httpResponse).bodyAsByteArray(); - Mockito.doReturn(new HttpHeaders()).when(httpResponse).headers(); - return httpResponse; + public void configurationPropertyReads() throws Exception { + DatabaseAccountManagerInternal databaseAccountManagerInternal = Mockito.mock(DatabaseAccountManagerInternal.class); + Mockito.when(databaseAccountManagerInternal.getDatabaseAccountFromEndpoint(Matchers.any())).thenReturn(Flux.just(new DatabaseAccount(GlobalEndPointManagerTest.dbAccountJson1))); + Mockito.when(databaseAccountManagerInternal.getServiceEndpoint()).thenReturn(new URI(TestConfigurations.HOST)); + GlobalEndpointManager globalEndpointManager = new GlobalEndpointManager(databaseAccountManagerInternal, new ConnectionPolicy(), new Configs()); + ReflectionUtils.setBackgroundRefreshLocationTimeIntervalInMS(globalEndpointManager, 1000); + globalEndpointManager.init(); + + GatewayServiceConfigurationReader configurationReader = new GatewayServiceConfigurationReader(globalEndpointManager); + assertThat(configurationReader.getDefaultConsistencyLevel()).isEqualTo(ConsistencyLevel.SESSION); + assertThat((boolean) configurationReader.getQueryEngineConfiguration().get("enableSpatialIndexing")).isTrue(); + assertThat(configurationReader.getSystemReplicationPolicy().getMaxReplicaSetSize()).isEqualTo(4); + assertThat(configurationReader.getUserReplicationPolicy().getMaxReplicaSetSize()).isEqualTo(4); + + Mockito.when(databaseAccountManagerInternal.getDatabaseAccountFromEndpoint(Matchers.any())).thenReturn(Flux.just(new DatabaseAccount(GlobalEndPointManagerTest.dbAccountJson2))); + Thread.sleep(2000); + assertThat(configurationReader.getDefaultConsistencyLevel()).isEqualTo(ConsistencyLevel.EVENTUAL); + assertThat((boolean) configurationReader.getQueryEngineConfiguration().get("enableSpatialIndexing")).isFalse(); + assertThat(configurationReader.getSystemReplicationPolicy().getMaxReplicaSetSize()).isEqualTo(5); + assertThat(configurationReader.getUserReplicationPolicy().getMaxReplicaSetSize()).isEqualTo(5); + + Mockito.when(databaseAccountManagerInternal.getDatabaseAccountFromEndpoint(Matchers.any())).thenReturn(Flux.just(new DatabaseAccount(GlobalEndPointManagerTest.dbAccountJson3))); + Thread.sleep(2000); + assertThat(configurationReader.getDefaultConsistencyLevel()).isEqualTo(ConsistencyLevel.SESSION); + assertThat((boolean) configurationReader.getQueryEngineConfiguration().get("enableSpatialIndexing")).isTrue(); + assertThat(configurationReader.getSystemReplicationPolicy().getMaxReplicaSetSize()).isEqualTo(4); + assertThat(configurationReader.getUserReplicationPolicy().getMaxReplicaSetSize()).isEqualTo(4); + + //Testing scenario of scheduled cache refresh with error + Mockito.when(databaseAccountManagerInternal.getDatabaseAccountFromEndpoint(Matchers.any())).thenReturn(Flux.error(BridgeInternal.createCosmosClientException(HttpConstants.StatusCodes.FORBIDDEN))); + Thread.sleep(2000); + assertThat(configurationReader.getDefaultConsistencyLevel()).isEqualTo(ConsistencyLevel.SESSION); + assertThat((boolean) configurationReader.getQueryEngineConfiguration().get("enableSpatialIndexing")).isTrue(); + assertThat(configurationReader.getSystemReplicationPolicy().getMaxReplicaSetSize()).isEqualTo(4); + assertThat(configurationReader.getUserReplicationPolicy().getMaxReplicaSetSize()).isEqualTo(4); } } diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfiguratorReaderMock.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfiguratorReaderMock.java index d7859b4a1b40..92c18eba0184 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfiguratorReaderMock.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GatewayServiceConfiguratorReaderMock.java @@ -47,7 +47,6 @@ public GatewayServiceConfiguratorReaderMock(ReplicationPolicy userReplicationPol ConsistencyLevel defaultConsistencyLevel) { this.gatewayServiceConfigurationReader = Mockito.mock(GatewayServiceConfigurationReader.class); - Mockito.doReturn(Mono.just(Mockito.mock(DatabaseAccount.class))).when(this.gatewayServiceConfigurationReader).initializeReaderAsync(); Mockito.doReturn(defaultConsistencyLevel).when(this.gatewayServiceConfigurationReader).getDefaultConsistencyLevel(); Mockito.doReturn(systemReplicationPolicy).when(this.gatewayServiceConfigurationReader).getSystemReplicationPolicy(); Mockito.doReturn(userReplicationPolicy).when(this.gatewayServiceConfigurationReader).getUserReplicationPolicy(); diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GlobalEndPointManagerTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GlobalEndPointManagerTest.java index 92ef88cf54b8..7fab9e5fd1d0 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GlobalEndPointManagerTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/GlobalEndPointManagerTest.java @@ -32,7 +32,7 @@ public class GlobalEndPointManagerTest { protected static final int TIMEOUT = 6000000; DatabaseAccountManagerInternal databaseAccountManagerInternal; - private String dbAccountJson1 = "{\"_self\":\"\",\"id\":\"testaccount\",\"_rid\":\"testaccount.documents.azure.com\",\"media\":\"//media/\",\"addresses\":\"//addresses/\"," + + static String dbAccountJson1 = "{\"_self\":\"\",\"id\":\"testaccount\",\"_rid\":\"testaccount.documents.azure.com\",\"media\":\"//media/\",\"addresses\":\"//addresses/\"," + "\"_dbs\":\"//dbs/\",\"writableLocations\":[{\"name\":\"East US\",\"databaseAccountEndpoint\":\"https://testaccount-eastus.documents.azure.com:443/\"}]," + "\"readableLocations\":[{\"name\":\"East US\",\"databaseAccountEndpoint\":\"https://testaccount-eastus.documents.azure.com:443/\"},{\"name\":\"East Asia\"," + "\"databaseAccountEndpoint\":\"https://testaccount-eastasia.documents.azure.com:443/\"}],\"enableMultipleWriteLocations\":false,\"userReplicationPolicy\":{\"asyncReplication\":false," + @@ -44,18 +44,18 @@ public class GlobalEndPointManagerTest { "\\\"maxSpatialQueryCells\\\":12,\\\"spatialMaxGeometryPointCount\\\":256,\\\"sqlAllowTop\\\":true,\\\"enableSpatialIndexing\\\":true}\"}\n"; - private String dbAccountJson2 = "{\"_self\":\"\",\"id\":\"testaccount\",\"_rid\":\"testaccount.documents.azure.com\",\"media\":\"//media/\"," + + static String dbAccountJson2 = "{\"_self\":\"\",\"id\":\"testaccount\",\"_rid\":\"testaccount.documents.azure.com\",\"media\":\"//media/\"," + "\"addresses\":\"//addresses/\",\"_dbs\":\"//dbs/\",\"writableLocations\":[{\"name\":\"East Asia\",\"databaseAccountEndpoint\":\"https://testaccount-eastasia.documents.azure" + ".com:443/\"}],\"readableLocations\":[{\"name\":\"East Asia\",\"databaseAccountEndpoint\":\"https://testaccount-eastasia.documents.azure.com:443/\"}]," + - "\"enableMultipleWriteLocations\":false,\"userReplicationPolicy\":{\"asyncReplication\":false,\"minReplicaSetSize\":3,\"maxReplicasetSize\":4}," + - "\"userConsistencyPolicy\":{\"defaultConsistencyLevel\":\"Session\"},\"systemReplicationPolicy\":{\"minReplicaSetSize\":3,\"maxReplicasetSize\":4}," + + "\"enableMultipleWriteLocations\":false,\"userReplicationPolicy\":{\"asyncReplication\":false,\"minReplicaSetSize\":3,\"maxReplicasetSize\":5}," + + "\"userConsistencyPolicy\":{\"defaultConsistencyLevel\":\"Eventual\"},\"systemReplicationPolicy\":{\"minReplicaSetSize\":3,\"maxReplicasetSize\":5}," + "\"readPolicy\":{\"primaryReadCoefficient\":1,\"secondaryReadCoefficient\":1},\"queryEngineConfiguration\":\"{\\\"maxSqlQueryInputLength\\\":262144,\\\"maxJoinsPerSqlQuery\\\":5," + "\\\"maxLogicalAndPerSqlQuery\\\":500,\\\"maxLogicalOrPerSqlQuery\\\":500,\\\"maxUdfRefPerSqlQuery\\\":10,\\\"maxInExpressionItemsCount\\\":16000," + "\\\"queryMaxInMemorySortDocumentCount\\\":500,\\\"maxQueryRequestTimeoutFraction\\\":0.9,\\\"sqlAllowNonFiniteNumbers\\\":false,\\\"sqlAllowAggregateFunctions\\\":true," + "\\\"sqlAllowSubQuery\\\":true,\\\"sqlAllowScalarSubQuery\\\":true,\\\"allowNewKeywords\\\":true,\\\"sqlAllowLike\\\":false,\\\"sqlAllowGroupByClause\\\":true," + - "\\\"maxSpatialQueryCells\\\":12,\\\"spatialMaxGeometryPointCount\\\":256,\\\"sqlAllowTop\\\":true,\\\"enableSpatialIndexing\\\":true}\"}"; + "\\\"maxSpatialQueryCells\\\":12,\\\"spatialMaxGeometryPointCount\\\":256,\\\"sqlAllowTop\\\":true,\\\"enableSpatialIndexing\\\":false}\"}"; - private String dbAccountJson3 = "{\"_self\":\"\",\"id\":\"testaccount\",\"_rid\":\"testaccount.documents.azure.com\",\"media\":\"//media/\"," + + static String dbAccountJson3 = "{\"_self\":\"\",\"id\":\"testaccount\",\"_rid\":\"testaccount.documents.azure.com\",\"media\":\"//media/\"," + "\"addresses\":\"//addresses/\",\"_dbs\":\"//dbs/\",\"writableLocations\":[{\"name\":\"West US\",\"databaseAccountEndpoint\":\"https://testaccount-westus.documents.azure" + ".com:443/\"}],\"readableLocations\":[{\"name\":\"West US\",\"databaseAccountEndpoint\":\"https://testaccount-westus.documents.azure.com:443/\"}]," + "\"enableMultipleWriteLocations\":false,\"userReplicationPolicy\":{\"asyncReplication\":false,\"minReplicaSetSize\":3,\"maxReplicasetSize\":4}," + @@ -66,7 +66,7 @@ public class GlobalEndPointManagerTest { "\\\"sqlAllowSubQuery\\\":true,\\\"sqlAllowScalarSubQuery\\\":true,\\\"allowNewKeywords\\\":true,\\\"sqlAllowLike\\\":false,\\\"sqlAllowGroupByClause\\\":true," + "\\\"maxSpatialQueryCells\\\":12,\\\"spatialMaxGeometryPointCount\\\":256,\\\"sqlAllowTop\\\":true,\\\"enableSpatialIndexing\\\":true}\"}"; - private String dbAccountJson4 = "{\"_self\":\"\",\"id\":\"testaccount\",\"_rid\":\"testaccount.documents.azure.com\",\"media\":\"//media/\",\"addresses\":\"//addresses/\"," + + static String dbAccountJson4 = "{\"_self\":\"\",\"id\":\"testaccount\",\"_rid\":\"testaccount.documents.azure.com\",\"media\":\"//media/\",\"addresses\":\"//addresses/\"," + "\"_dbs\":\"//dbs/\",\"writableLocations\":[{\"name\":\"East US\",\"databaseAccountEndpoint\":\"https://testaccount-eastus.documents.azure.com:443/\"},{\"name\":\"East Asia\"," + "\"databaseAccountEndpoint\":\"https://testaccount-eastasia.documents.azure.com:443/\"}],\"readableLocations\":[{\"name\":\"East US\"," + "\"databaseAccountEndpoint\":\"https://testaccount-eastus.documents.azure.com:443/\"},{\"name\":\"East Asia\",\"databaseAccountEndpoint\":\"https://testaccount-eastasia.documents" + diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/ReflectionUtils.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/ReflectionUtils.java index 413b986cb0ef..f78e7000fedb 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/ReflectionUtils.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/directconnectivity/ReflectionUtils.java @@ -3,11 +3,11 @@ package com.azure.cosmos.implementation.directconnectivity; -import com.azure.cosmos.BridgeInternal; import com.azure.cosmos.CosmosAsyncClient; import com.azure.cosmos.CosmosBridgeInternal; import com.azure.cosmos.CosmosClient; import com.azure.cosmos.implementation.AsyncDocumentClient; +import com.azure.cosmos.implementation.GlobalEndpointManager; import com.azure.cosmos.implementation.RxDocumentClientImpl; import com.azure.cosmos.implementation.http.HttpClient; import org.apache.commons.lang3.reflect.FieldUtils; @@ -94,4 +94,12 @@ public static AsyncDocumentClient getAsyncDocumentClient(CosmosAsyncClient clien public static void setAsyncDocumentClient(CosmosAsyncClient client, RxDocumentClientImpl rxClient) { set(client, rxClient, "asyncDocumentClient"); } + + public static GatewayServiceConfigurationReader getServiceConfigurationReader(RxDocumentClientImpl rxDocumentClient){ + return get(GatewayServiceConfigurationReader.class, rxDocumentClient, "gatewayConfigurationReader"); + } + + public static void setBackgroundRefreshLocationTimeIntervalInMS(GlobalEndpointManager globalEndPointManager, int millSec){ + set(globalEndPointManager, millSec, "backgroundRefreshLocationTimeIntervalInMS"); + } } diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TokenResolverTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TokenResolverTest.java index 4a20c4b7571d..c01cce8c3bc1 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TokenResolverTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TokenResolverTest.java @@ -507,7 +507,9 @@ private Permission getPermission(Resource resource, String permissionId, Permiss private TokenResolver getTokenResolver(PermissionMode permissionMode) { return (RequestVerb requestVerb, String resourceIdOrFullName, CosmosResourceType resourceType, Map properties) -> { - if (permissionMode == null) { + if(resourceType.equals(CosmosResourceType.System)) { + return readPermission.getToken(); + } if (permissionMode == null) { return "invalid"; } else if (permissionMode.equals(PermissionMode.READ)) { return readPermission.getToken();