Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fixed dont retry writes on forbidden #307

Merged
merged 2 commits into from
Jan 14, 2020
Merged

Fixed dont retry writes on forbidden #307

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b7ff082
Fixed dont retry writes on forbidden
kushagraThapar Jan 14, 2020
a22f31d
Replaced this.isReadRequest with isReadRequest
kushagraThapar Jan 14, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 36 additions & 36 deletions gateway/src/main/java/com/microsoft/azure/cosmosdb/rx/internal/ClientRetryPolicy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,12 +99,6 @@ public Single<ShouldRetryResult> 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, false, e);
}

this.retryContext = null;
// Received 403.3 on write region, initiate the endpoint re-discovery
DocumentClientException clientException = Utils.as(e, DocumentClientException.class);
Expand All @@ -115,8 +109,8 @@ public Single<ShouldRetryResult> 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, 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)
Expand All @@ -125,14 +119,18 @@ public Single<ShouldRetryResult> shouldRetry(Exception e) {
Exceptions.isSubStatusCode(clientException, HttpConstants.SubStatusCodes.DATABASE_ACCOUNT_NOTFOUND) &&
(this.isReadRequest || this.canUseMultipleWriteLocations))
{
logger.warn("Endpoint not available for reads. Will refresh cache and retry. {}", e.toString());
return this.shouldRetryOnEndpointFailureAsync(true, false, e);
logger.warn("Endpoint not available for reads. Will refresh cache and retry. ", e);
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, 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 &&
Expand Down Expand Up @@ -175,29 +173,20 @@ private ShouldRetryResult shouldRetryOnSessionNotAvailable() {
}
}

private Single<ShouldRetryResult> shouldRetryOnEndpointFailureAsync(boolean isReadRequest , boolean forceRefresh, Exception e) {
private Single<ShouldRetryResult> shouldRetryOnEndpointFailureAsync(boolean isReadRequest , boolean forceRefresh) {
if (!this.enableEndpointDiscovery || this.failoverRetryCount > MaxRetryCount) {
logger.warn("ShouldRetryOnEndpointFailureAsync() Not retrying. Retry count = {}", this.failoverRetryCount);
return Single.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);
}
Completable refreshCompletion = 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.
// 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.
Expand All @@ -206,22 +195,33 @@ private Single<ShouldRetryResult> shouldRetryOnEndpointFailureAsync(boolean isRe
} else {
retryDelay = Duration.ofMillis(ClientRetryPolicy.RetryIntervalInMS);
}
this.retryContext = new RetryContext(this.failoverRetryCount, false);
return refreshCompletion.andThen(Single.just(ShouldRetryResult.retryAfter(retryDelay)));
}

private Single<ShouldRetryResult> shouldNotRetryOnEndpointFailureAsync(boolean isReadRequest , boolean forceRefresh) {
if (!this.enableEndpointDiscovery || this.failoverRetryCount > MaxRetryCount) {
logger.warn("ShouldRetryOnEndpointFailureAsync() Not retrying. Retry count = {}", this.failoverRetryCount);
return Single.just(ShouldRetryResult.noRetry());
}
Completable refreshCompletion = this.refreshLocation(isReadRequest, forceRefresh);
return refreshCompletion.andThen(Single.just(ShouldRetryResult.noRetry()));
}

Completable refreshCompletion = this.globalEndpointManager.refreshLocationAsync(null, forceRefresh);
private Completable refreshLocation(boolean isReadRequest, boolean forceRefresh) {
this.failoverRetryCount++;

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)));
// Mark the current read endpoint as unavailable
if (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 refreshCompletion
.andThen(Single.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
Expand Down