diff --git a/gateway/src/main/java/com/microsoft/azure/cosmosdb/rx/internal/ClientRetryPolicy.java b/gateway/src/main/java/com/microsoft/azure/cosmosdb/rx/internal/ClientRetryPolicy.java index edfc56ac9..75efb0083 100644 --- a/gateway/src/main/java/com/microsoft/azure/cosmosdb/rx/internal/ClientRetryPolicy.java +++ b/gateway/src/main/java/com/microsoft/azure/cosmosdb/rx/internal/ClientRetryPolicy.java @@ -30,6 +30,7 @@ import org.apache.commons.collections4.list.UnmodifiableList; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import rx.Completable; import rx.Single; import java.net.URL; @@ -98,6 +99,12 @@ public Single shouldRetry(Exception e) { return rxNettyConnectionPoolExhaustedRetry.shouldRetry(e); } + // 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, e); + } + this.retryContext = null; // Received 403.3 on write region, initiate the endpoint re-discovery DocumentClientException clientException = Utils.as(e, DocumentClientException.class); @@ -109,7 +116,7 @@ public Single shouldRetry(Exception e) { Exceptions.isSubStatusCode(clientException, HttpConstants.SubStatusCodes.FORBIDDEN_WRITEFORBIDDEN)) { logger.warn("Endpoint not writable. Will refresh cache and retry. {}", e.toString()); - return this.shouldRetryOnEndpointFailureAsync(false); + return this.shouldRetryOnEndpointFailureAsync(false, e); } // Regional endpoint is not available yet for reads (e.g. add/ online of region is in progress) @@ -119,13 +126,7 @@ public Single shouldRetry(Exception e) { (this.isReadRequest || this.canUseMultipleWriteLocations)) { logger.warn("Endpoint not available for reads. Will refresh cache and retry. {}", e.toString()); - return this.shouldRetryOnEndpointFailureAsync(true); - } - - // 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); + return this.shouldRetryOnEndpointFailureAsync(true, e); } if (clientException != null && @@ -168,7 +169,7 @@ private ShouldRetryResult shouldRetryOnSessionNotAvailable() { } } - private Single shouldRetryOnEndpointFailureAsync(boolean isReadRequest) { + private Single shouldRetryOnEndpointFailureAsync(boolean isReadRequest, Exception e) { if (!this.enableEndpointDiscovery || this.failoverRetryCount > MaxRetryCount) { logger.warn("ShouldRetryOnEndpointFailureAsync() Not retrying. Retry count = {}", this.failoverRetryCount); return Single.just(ShouldRetryResult.noRetry()); @@ -200,8 +201,21 @@ private Single shouldRetryOnEndpointFailureAsync(boolean isRe retryDelay = Duration.ofMillis(ClientRetryPolicy.RetryIntervalInMS); } this.retryContext = new RetryContext(this.failoverRetryCount, false); - return this.globalEndpointManager.refreshLocationAsync(null) - .andThen(Single.just(ShouldRetryResult.retryAfter(retryDelay))); + + Completable refreshCompletion = this.globalEndpointManager.refreshLocationAsync(null); + + 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 refreshCompletion + .andThen(Single.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 refreshCompletion + .andThen(Single.just(ShouldRetryResult.noRetry())); + } } @Override diff --git a/gateway/src/test/java/com/microsoft/azure/cosmosdb/rx/internal/ClientRetryPolicyTest.java b/gateway/src/test/java/com/microsoft/azure/cosmosdb/rx/internal/ClientRetryPolicyTest.java index b6c6157fb..7f007c771 100644 --- a/gateway/src/test/java/com/microsoft/azure/cosmosdb/rx/internal/ClientRetryPolicyTest.java +++ b/gateway/src/test/java/com/microsoft/azure/cosmosdb/rx/internal/ClientRetryPolicyTest.java @@ -87,10 +87,10 @@ public void networkFailureOnWrite() throws Exception { clientRetryPolicy.onBeforeSendRequest(dsr); for (int i = 0; i < 10; i++) { Single shouldRetry = clientRetryPolicy.shouldRetry(exception); + // We don't want to retry writes on network failure 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()); @@ -98,6 +98,62 @@ 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 URL("http://localhost")).when(endpointManager).resolveServiceEndpoint(Mockito.any(RxDocumentServiceRequest.class)); + Mockito.doReturn(Completable.complete()).when(endpointManager).refreshLocationAsync(Mockito.eq(null)); + 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++) { + Single shouldRetry = clientRetryPolicy.shouldRetry(exception); + // We don't want to retry writes on network failure + 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 URL("http://localhost")).when(endpointManager).resolveServiceEndpoint(Mockito.any(RxDocumentServiceRequest.class)); + Mockito.doReturn(Completable.complete()).when(endpointManager).refreshLocationAsync(Mockito.eq(null)); + 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++) { + Single shouldRetry = clientRetryPolicy.shouldRetry(exception); + // We don't want to retry writes on network failure + 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();