From 5be6b4066ad42c1618b7e10f3e387fbeea8823f1 Mon Sep 17 00:00:00 2001 From: Kushagra Thapar Date: Tue, 7 Jan 2020 12:55:51 -0800 Subject: [PATCH 1/3] Don't retry on network failure on writes --- .../implementation/ClientRetryPolicy.java | 35 +++++++----- .../implementation/ClientRetryPolicyTest.java | 57 ++++++++++++++++++- 2 files changed, 77 insertions(+), 15 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java index 105a3da30e8f..48e30bdfb0bd 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java @@ -71,6 +71,12 @@ public Mono shouldRetry(Exception e) { return Mono.just(ShouldRetryResult.error(e)); } + // Received Connection error (HttpException), initiate the endpoint rediscovery + if (WebExceptionUtility.isNetworkFailure(e)) { + logger.warn("Endpoint not reachable. Will refresh cache and retry. " , e); + return this.shouldRetryOnEndpointFailureAsync(this.isReadRequest, false, e); + } + this.retryContext = null; // Received 403.3 on write region, initiate the endpoint re-discovery CosmosClientException clientException = Utils.as(e, CosmosClientException.class); @@ -81,8 +87,8 @@ public Mono shouldRetry(Exception e) { Exceptions.isStatusCode(clientException, HttpConstants.StatusCodes.FORBIDDEN) && Exceptions.isSubStatusCode(clientException, HttpConstants.SubStatusCodes.FORBIDDEN_WRITEFORBIDDEN)) { - logger.warn("Endpoint not writable. Will refresh cache and retry. {}", e.toString()); - return this.shouldRetryOnEndpointFailureAsync(false, true); + logger.warn("Endpoint not writable. Will refresh cache and retry. ", e); + return this.shouldRetryOnEndpointFailureAsync(false, true, e); } // Regional endpoint is not available yet for reads (e.g. add/ online of region is in progress) @@ -91,14 +97,8 @@ public Mono shouldRetry(Exception e) { Exceptions.isSubStatusCode(clientException, HttpConstants.SubStatusCodes.DATABASE_ACCOUNT_NOTFOUND) && this.isReadRequest) { - logger.warn("Endpoint not available for reads. Will refresh cache and retry. {}", e.toString()); - return this.shouldRetryOnEndpointFailureAsync(true, false); - } - - // Received Connection error (HttpRequestException), initiate the endpoint rediscovery - if (WebExceptionUtility.isNetworkFailure(e)) { - logger.warn("Endpoint not reachable. Will refresh cache and retry. {}" , e.toString()); - return this.shouldRetryOnEndpointFailureAsync(this.isReadRequest, false); + logger.warn("Endpoint not available for reads. Will refresh cache and retry. ", e); + return this.shouldRetryOnEndpointFailureAsync(true, false, e); } if (clientException != null && @@ -141,7 +141,7 @@ private ShouldRetryResult shouldRetryOnSessionNotAvailable() { } } - private Mono shouldRetryOnEndpointFailureAsync(boolean isReadRequest , boolean forceRefresh) { + private Mono shouldRetryOnEndpointFailureAsync(boolean isReadRequest , boolean forceRefresh, Exception e) { if (!this.enableEndpointDiscovery || this.failoverRetryCount > MaxRetryCount) { logger.warn("ShouldRetryOnEndpointFailureAsync() Not retrying. Retry count = {}", this.failoverRetryCount); return Mono.just(ShouldRetryResult.noRetry()); @@ -173,8 +173,17 @@ private Mono shouldRetryOnEndpointFailureAsync(boolean isRead retryDelay = Duration.ofMillis(ClientRetryPolicy.RetryIntervalInMS); } this.retryContext = new RetryContext(this.failoverRetryCount, false); - return this.globalEndpointManager.refreshLocationAsync(null, forceRefresh) - .then(Mono.just(ShouldRetryResult.retryAfter(retryDelay))); + Mono completable = this.globalEndpointManager.refreshLocationAsync(null, forceRefresh); + if (isReadRequest || WebExceptionUtility.isWebExceptionRetriable(e)) { + // refresh cache and + // if it is a read request or if it is a write but we are sure the write + // hasn't reached the service retry + return completable.then(Mono.just(ShouldRetryResult.retryAfter(retryDelay))); + } else { + // refresh cache and + // no retry for writes which we are not sure if have reached to the service or not + return completable.then(Mono.just(ShouldRetryResult.noRetry())); + } } @Override diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/ClientRetryPolicyTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/ClientRetryPolicyTest.java index 2311c8565ad0..2ca91c2df27a 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/ClientRetryPolicyTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/implementation/ClientRetryPolicyTest.java @@ -66,8 +66,7 @@ public void networkFailureOnWrite() throws Exception { Mono shouldRetry = clientRetryPolicy.shouldRetry(exception); validateSuccess(shouldRetry, ShouldRetryValidator.builder() .nullException() - .shouldRetry(true) - .backOfTime(i > 0 ? Duration.ofMillis(ClientRetryPolicy.RetryIntervalInMS) : Duration.ZERO) + .shouldRetry(false) .build()); Mockito.verify(endpointManager, Mockito.times(0)).markEndpointUnavailableForRead(Mockito.any()); @@ -75,6 +74,60 @@ public void networkFailureOnWrite() throws Exception { } } + @Test(groups = "unit") + public void networkFailureOnUpsert() throws Exception { + RetryOptions retryOptions = new RetryOptions(); + GlobalEndpointManager endpointManager = Mockito.mock(GlobalEndpointManager.class); + Mockito.doReturn(new URI("http://localhost")).when(endpointManager).resolveServiceEndpoint(Mockito.any(RxDocumentServiceRequest.class)); + Mockito.doReturn(Mono.empty()).when(endpointManager).refreshLocationAsync(Mockito.eq(null), Mockito.eq(false)); + ClientRetryPolicy clientRetryPolicy = new ClientRetryPolicy(endpointManager, true, retryOptions); + + Exception exception = ReadTimeoutException.INSTANCE; + + RxDocumentServiceRequest dsr = RxDocumentServiceRequest.createFromName( + OperationType.Upsert, "/dbs/db/colls/col/docs/docId", ResourceType.Document); + dsr.requestContext = Mockito.mock(DocumentServiceRequestContext.class); + + clientRetryPolicy.onBeforeSendRequest(dsr); + for (int i = 0; i < 10; i++) { + Mono shouldRetry = clientRetryPolicy.shouldRetry(exception); + validateSuccess(shouldRetry, ShouldRetryValidator.builder() + .nullException() + .shouldRetry(false) + .build()); + + Mockito.verify(endpointManager, Mockito.times(0)).markEndpointUnavailableForRead(Mockito.any()); + Mockito.verify(endpointManager, Mockito.times(i + 1)).markEndpointUnavailableForWrite(Mockito.any()); + } + } + + @Test(groups = "unit") + public void networkFailureOnDelete() throws Exception { + RetryOptions retryOptions = new RetryOptions(); + GlobalEndpointManager endpointManager = Mockito.mock(GlobalEndpointManager.class); + Mockito.doReturn(new URI("http://localhost")).when(endpointManager).resolveServiceEndpoint(Mockito.any(RxDocumentServiceRequest.class)); + Mockito.doReturn(Mono.empty()).when(endpointManager).refreshLocationAsync(Mockito.eq(null), Mockito.eq(false)); + ClientRetryPolicy clientRetryPolicy = new ClientRetryPolicy(endpointManager, true, retryOptions); + + Exception exception = ReadTimeoutException.INSTANCE; + + RxDocumentServiceRequest dsr = RxDocumentServiceRequest.createFromName( + OperationType.Delete, "/dbs/db/colls/col/docs/docId", ResourceType.Document); + dsr.requestContext = Mockito.mock(DocumentServiceRequestContext.class); + + clientRetryPolicy.onBeforeSendRequest(dsr); + for (int i = 0; i < 10; i++) { + Mono shouldRetry = clientRetryPolicy.shouldRetry(exception); + validateSuccess(shouldRetry, ShouldRetryValidator.builder() + .nullException() + .shouldRetry(false) + .build()); + + Mockito.verify(endpointManager, Mockito.times(0)).markEndpointUnavailableForRead(Mockito.any()); + Mockito.verify(endpointManager, Mockito.times(i + 1)).markEndpointUnavailableForWrite(Mockito.any()); + } + } + @Test(groups = "unit") public void onBeforeSendRequestNotInvoked() { RetryOptions retryOptions = new RetryOptions(); From bdcc8e771003200e1a4ca55e2e638178391f9dca Mon Sep 17 00:00:00 2001 From: Kushagra Thapar Date: Mon, 13 Jan 2020 14:46:15 -0800 Subject: [PATCH 2/3] Fixing the logic to retry on forbidden status code --- .../implementation/ClientRetryPolicy.java | 69 +++++++++++-------- 1 file changed, 39 insertions(+), 30 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java index 48e30bdfb0bd..32e6d1561a2b 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java @@ -71,12 +71,6 @@ public Mono shouldRetry(Exception e) { return Mono.just(ShouldRetryResult.error(e)); } - // Received Connection error (HttpException), initiate the endpoint rediscovery - if (WebExceptionUtility.isNetworkFailure(e)) { - logger.warn("Endpoint not reachable. Will refresh cache and retry. " , e); - return this.shouldRetryOnEndpointFailureAsync(this.isReadRequest, false, e); - } - this.retryContext = null; // Received 403.3 on write region, initiate the endpoint re-discovery CosmosClientException clientException = Utils.as(e, CosmosClientException.class); @@ -87,8 +81,8 @@ public Mono shouldRetry(Exception e) { Exceptions.isStatusCode(clientException, HttpConstants.StatusCodes.FORBIDDEN) && Exceptions.isSubStatusCode(clientException, HttpConstants.SubStatusCodes.FORBIDDEN_WRITEFORBIDDEN)) { - logger.warn("Endpoint not writable. Will refresh cache and retry. ", e); - return this.shouldRetryOnEndpointFailureAsync(false, true, e); + logger.warn("Endpoint not writable. Will refresh cache and retry ", e); + return this.shouldRetryOnEndpointFailureAsync(false, true); } // Regional endpoint is not available yet for reads (e.g. add/ online of region is in progress) @@ -98,7 +92,17 @@ public Mono shouldRetry(Exception e) { this.isReadRequest) { logger.warn("Endpoint not available for reads. Will refresh cache and retry. ", e); - return this.shouldRetryOnEndpointFailureAsync(true, false, e); + return this.shouldRetryOnEndpointFailureAsync(true, false); + } + + // Received Connection error (HttpException), initiate the endpoint rediscovery + if (WebExceptionUtility.isNetworkFailure(e)) { + if (this.isReadRequest || WebExceptionUtility.isWebExceptionRetriable(e)) { + logger.warn("Endpoint not reachable. Will refresh cache and retry. ", e); + return this.shouldRetryOnEndpointFailureAsync(this.isReadRequest, false); + } else { + return this.shouldNotRetryOnEndpointFailureAsync(this.isReadRequest, false); + } } if (clientException != null && @@ -141,22 +145,13 @@ private ShouldRetryResult shouldRetryOnSessionNotAvailable() { } } - private Mono shouldRetryOnEndpointFailureAsync(boolean isReadRequest , boolean forceRefresh, Exception e) { + private Mono shouldRetryOnEndpointFailureAsync(boolean isReadRequest , boolean forceRefresh) { if (!this.enableEndpointDiscovery || this.failoverRetryCount > MaxRetryCount) { logger.warn("ShouldRetryOnEndpointFailureAsync() Not retrying. Retry count = {}", this.failoverRetryCount); return Mono.just(ShouldRetryResult.noRetry()); } - this.failoverRetryCount++; - - // Mark the current read endpoint as unavailable - if (this.isReadRequest) { - logger.warn("marking the endpoint {} as unavailable for read",this.locationEndpoint); - this.globalEndpointManager.markEndpointUnavailableForRead(this.locationEndpoint); - } else { - logger.warn("marking the endpoint {} as unavailable for write",this.locationEndpoint); - this.globalEndpointManager.markEndpointUnavailableForWrite(this.locationEndpoint); - } + Mono refreshLocationCompletable = this.refreshLocation(isReadRequest, forceRefresh); // Some requests may be in progress when the endpoint manager and client are closed. // In that case, the request won't succeed since the http client is closed. @@ -172,18 +167,32 @@ private Mono shouldRetryOnEndpointFailureAsync(boolean isRead } else { retryDelay = Duration.ofMillis(ClientRetryPolicy.RetryIntervalInMS); } - this.retryContext = new RetryContext(this.failoverRetryCount, false); - Mono completable = this.globalEndpointManager.refreshLocationAsync(null, forceRefresh); - if (isReadRequest || WebExceptionUtility.isWebExceptionRetriable(e)) { - // refresh cache and - // if it is a read request or if it is a write but we are sure the write - // hasn't reached the service retry - return completable.then(Mono.just(ShouldRetryResult.retryAfter(retryDelay))); + return refreshLocationCompletable.then(Mono.just(ShouldRetryResult.retryAfter(retryDelay))); + } + + private Mono shouldNotRetryOnEndpointFailureAsync(boolean isReadRequest , boolean forceRefresh) { + if (!this.enableEndpointDiscovery || this.failoverRetryCount > MaxRetryCount) { + logger.warn("ShouldRetryOnEndpointFailureAsync() Not retrying. Retry count = {}", this.failoverRetryCount); + return Mono.just(ShouldRetryResult.noRetry()); + } + Mono refreshLocationCompletable = this.refreshLocation(isReadRequest, forceRefresh); + return refreshLocationCompletable.then(Mono.just(ShouldRetryResult.noRetry())); + } + + private Mono refreshLocation(boolean isReadRequest, boolean forceRefresh) { + this.failoverRetryCount++; + + // Mark the current read endpoint as unavailable + if (this.isReadRequest) { + logger.warn("marking the endpoint {} as unavailable for read",this.locationEndpoint); + this.globalEndpointManager.markEndpointUnavailableForRead(this.locationEndpoint); } else { - // refresh cache and - // no retry for writes which we are not sure if have reached to the service or not - return completable.then(Mono.just(ShouldRetryResult.noRetry())); + logger.warn("marking the endpoint {} as unavailable for write",this.locationEndpoint); + this.globalEndpointManager.markEndpointUnavailableForWrite(this.locationEndpoint); } + + this.retryContext = new RetryContext(this.failoverRetryCount, false); + return this.globalEndpointManager.refreshLocationAsync(null, forceRefresh); } @Override From 69552e653dd1c68872f632fd15845aa11ad0793f Mon Sep 17 00:00:00 2001 From: Kushagra Thapar Date: Mon, 13 Jan 2020 15:11:07 -0800 Subject: [PATCH 3/3] Replaced this.isReadRequest with local isReadRequest --- .../com/azure/cosmos/implementation/ClientRetryPolicy.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java index 32e6d1561a2b..13ea67b4d0e3 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ClientRetryPolicy.java @@ -158,7 +158,7 @@ private Mono shouldRetryOnEndpointFailureAsync(boolean isRead // Therefore just skip the retry here to avoid the delay because retrying won't go through in the end. Duration retryDelay = Duration.ZERO; - if (!this.isReadRequest) { + if (!isReadRequest) { logger.debug("Failover happening. retryCount {}", this.failoverRetryCount); if (this.failoverRetryCount > 1) { //if retried both endpoints, follow regular retry interval. @@ -183,7 +183,7 @@ private Mono refreshLocation(boolean isReadRequest, boolean forceRefresh) this.failoverRetryCount++; // Mark the current read endpoint as unavailable - if (this.isReadRequest) { + if (isReadRequest) { logger.warn("marking the endpoint {} as unavailable for read",this.locationEndpoint); this.globalEndpointManager.markEndpointUnavailableForRead(this.locationEndpoint); } else {