Azure
/
azure-cosmosdb-java
зеркало из https://github.com/Azure/azure-cosmosdb-java.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность

DocumentClientException is now thread-safe (#323) 

* DocumentClientException is now thread-safe (but offers no consistency guarantees) and we have a test that request/response headers can be enumerated while they're being updated (sans ConcurrentModificationExceptions).

* Bug fix: Adjust for fact that ConcurrentMap does not accept the null values that we sometimes pass.

* Added two new DocumentClientExceptionTest methods and increased the timeout interval on ConsistencyTest2.validateSessionTokenAsync because it times out regularly in local and CI test runs with the old timeout interval.
This commit is contained in:
David Noble 2020-03-11 16:29:45 -07:00 коммит произвёл GitHub
Родитель 3511f32e69
Коммит ca47cadbc6
Не найден ключ, соответствующий данной подписи
Идентификатор ключа GPG: 4AEE18F83AFDEB23
7 изменённых файлов: 324 добавлений и 117 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

8
commons/pom.xml
Просмотреть файл

@ -188,6 +188,12 @@ SOFTWARE.
<artifactId>guava</artifactId>
<version>${guava.version}</version>
<scope>test</scope>
</dependency>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-core</artifactId>
<version>${micrometer.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

19
commons/src/main/java/com/microsoft/azure/cosmosdb/BridgeInternal.java
Просмотреть файл

@ -31,7 +31,6 @@ import com.microsoft.azure.cosmosdb.internal.query.metrics.ClientSideMetrics;
import com.microsoft.azure.cosmosdb.rx.internal.RxDocumentServiceResponse;
import com.microsoft.azure.cosmosdb.rx.internal.Strings;
import java.net.URI;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@ -179,29 +178,29 @@ public class BridgeInternal {
}
public static <E extends DocumentClientException> E setResourceAddress(E e, String resourceAddress) {
e.resourceAddress = resourceAddress;
e.setResourceAddress(resourceAddress);
return e;
}
public static <E extends DocumentClientException> long getLSN(E e) {
return e.lsn;
return e.getLsn();
}
public static <E extends DocumentClientException> String getPartitionKeyRangeId(E e) {
return e.partitionKeyRangeId;
return e.getPartitionKeyRangeId();
}
public static <E extends DocumentClientException> String getResourceAddress(E e) {
return e.resourceAddress;
return e.getResourceAddress();
}
public static <E extends DocumentClientException> E setLSN(E e, long lsn) {
e.lsn = lsn;
e.setLsn(lsn);
return e;
}
public static <E extends DocumentClientException> E setPartitionKeyRangeId(E e, String partitionKeyRangeId) {
e.partitionKeyRangeId = partitionKeyRangeId;
e.setPartitionKeyRangeId(partitionKeyRangeId);
return e;
}
@ -218,15 +217,15 @@ public class BridgeInternal {
}
public static <E extends DocumentClientException> Uri getRequestUri(DocumentClientException documentClientException) {
return documentClientException.requestUri;
return documentClientException.getRequestUri();
}
public static <E extends DocumentClientException> void setRequestHeaders(DocumentClientException documentClientException, Map<String, String> requestHeaders) {
documentClientException.requestHeaders = requestHeaders;
documentClientException.setRequestHeaders(requestHeaders);
}
public static <E extends DocumentClientException> Map<String, String> getRequestHeaders(DocumentClientException documentClientException) {
return documentClientException.requestHeaders;
return documentClientException.getRequestHeaders();
}
public static Map<String, Object> getQueryEngineConfiuration(DatabaseAccount databaseAccount) {

269
commons/src/main/java/com/microsoft/azure/cosmosdb/DocumentClientException.java
Просмотреть файл

@ -23,15 +23,14 @@
package com.microsoft.azure.cosmosdb;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;
import com.microsoft.azure.cosmosdb.internal.Constants;
import com.microsoft.azure.cosmosdb.internal.HttpConstants;
import com.microsoft.azure.cosmosdb.internal.directconnectivity.Uri;
import org.apache.commons.lang3.StringUtils;
import com.microsoft.azure.cosmosdb.internal.Constants;
import com.microsoft.azure.cosmosdb.internal.HttpConstants;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* This class defines a custom exception type for all operations on
@ -51,25 +50,19 @@ public class DocumentClientException extends Exception {
private static final long serialVersionUID = 1L;
private final int statusCode;
private final Map<String, String> requestHeaders;
private final Map<String, String> responseHeaders;
private final int statusCode;
private ClientSideRequestStatistics clientSideRequestStatistics;
private Error error;
long lsn;
String partitionKeyRangeId;
Map<String, String> requestHeaders;
Uri requestUri;
String resourceAddress;
private DocumentClientException(int statusCode, String message, Map<String, String> responseHeaders, Throwable cause) {
super(message, cause);
this.statusCode = statusCode;
this.responseHeaders = responseHeaders == null ? new HashMap<>() : new HashMap<>(responseHeaders);
}
private volatile ClientSideRequestStatistics clientSideRequestStatistics;
private volatile Error error;
private volatile long lsn;
private volatile String partitionKeyRangeId;
private volatile Uri requestUri;
private volatile String resourceAddress;
/**
* Creates a new instance of the DocumentClientException class.
* Creates a new instance of the {@link DocumentClientException} class.
*
* @param statusCode the http status code of the response.
*/
@ -78,7 +71,7 @@ public class DocumentClientException extends Exception {
}
/**
* Creates a new instance of the DocumentClientException class.
* Creates a new instance of the {@link DocumentClientException} class.
*
* @param statusCode the http status code of the response.
* @param errorMessage the error message.
@ -90,7 +83,7 @@ public class DocumentClientException extends Exception {
}
/**
* Creates a new instance of the DocumentClientException class.
* Creates a new instance of the {@link DocumentClientException} class.
*
* @param statusCode the http status code of the response.
* @param innerException the original exception.
@ -100,7 +93,7 @@ public class DocumentClientException extends Exception {
}
/**
* Creates a new instance of the DocumentClientException class.
* Creates a new instance of the {@link DocumentClientException} class.
*
* @param statusCode the http status code of the response.
* @param errorResource the error resource object.
@ -111,7 +104,7 @@ public class DocumentClientException extends Exception {
}
/**
* Creates a new instance of the DocumentClientException class.
* Creates a new instance of the {@link DocumentClientException} class.
*
* @param resourceAddress the address of the resource the request is associated with.
* @param statusCode the http status code of the response.
@ -119,30 +112,40 @@ public class DocumentClientException extends Exception {
* @param responseHeaders the response headers.
*/
public DocumentClientException(String resourceAddress, int statusCode, Error errorResource, Map<String, String> responseHeaders) {
public DocumentClientException(
String resourceAddress, int statusCode, Error errorResource, Map<String, String> responseHeaders) {
this(statusCode, errorResource == null ? null : errorResource.getMessage(), responseHeaders, null);
this.resourceAddress = resourceAddress;
this.error = errorResource;
}
/** Creates a new instance of the DocumentClientException class.
/**
* Creates a new instance of the {@link DocumentClientException} class.
*
* @param message the string message.
* @param statusCode the http status code of the response.
* @param exception the exception object.
* @param responseHeaders the response headers.
* @param resourceAddress the address of the resource the request is associated with.
*/
public DocumentClientException(String message, Exception exception, Map<String, String> responseHeaders, int statusCode, String resourceAddress) {
public DocumentClientException(
String message,
Exception exception,
Map<String, String> responseHeaders,
int statusCode,
String resourceAddress) {
this(statusCode, message, responseHeaders, exception);
this.resourceAddress = resourceAddress;
}
@Override
public String getMessage() {
if (clientSideRequestStatistics == null) {
return getInnerErrorMessage();
}
return getInnerErrorMessage() + ", " + clientSideRequestStatistics.toString();
private DocumentClientException(
int statusCode, String message, Map<String, String> responseHeaders, Throwable cause) {
super(message, cause);
this.requestHeaders = new ConcurrentHashMap<>();
this.responseHeaders = responseHeaders == null
? new ConcurrentHashMap<>()
: new ConcurrentHashMap<>(responseHeaders);
this.statusCode = statusCode;
}
/**
@ -158,6 +161,123 @@ public class DocumentClientException extends Exception {
return null;
}
/**
* Gets the Client side request statistics associated with this exception.
*
* @return Client side request statistics associated with this exception.
*/
public ClientSideRequestStatistics getClientSideRequestStatistics() {
return clientSideRequestStatistics;
}
/**
* Sets the Client side request statistics associated with this exception.
*/
// TODO (DANOBLE) Consider changing the access of this method to package private and adding it to BridgeInternal
public void setClientSideRequestStatistics(ClientSideRequestStatistics clientSideRequestStatistics) {
this.clientSideRequestStatistics = clientSideRequestStatistics;
}
/**
* Gets the error code associated with the exception.
*
* @return the error.
*/
public Error getError() {
return this.error;
}
/**
* Gets the LSN associated with this {@link DocumentClientException exception}.
*
* @return the LSN associated with this {@link DocumentClientException exception}.
*/
public long getLsn() {
return lsn;
}
void setLsn(long lsn) {
this.lsn = lsn;
}
@Override
public String getMessage() {
if (clientSideRequestStatistics == null) {
return getInnerErrorMessage();
}
return getInnerErrorMessage() + ", " + clientSideRequestStatistics.toString();
}
public String getPartitionKeyRangeId() {
return this.partitionKeyRangeId;
}
void setPartitionKeyRangeId(String partitionKeyRangeId) {
this.partitionKeyRangeId = partitionKeyRangeId;
}
public Map<String, String> getRequestHeaders() {
return this.requestHeaders;
}
void setRequestHeaders(Map<String, String> values) {
this.requestHeaders.clear();
if (values != null) {
for (Map.Entry<String, String> entry : values.entrySet()) {
final String key = entry.getKey();
final String value = entry.getValue();
if (key != null && value != null) {
this.requestHeaders.put(key, value);
}
}
}
}
public Uri getRequestUri() {
return requestUri;
}
public void setRequestUri(Uri requestUri) {
this.requestUri = requestUri;
}
/**
* Gets the response headers as key-value pairs
*
* @return the response headers
*/
public Map<String, String> getResponseHeaders() {
return this.responseHeaders;
}
/**
* Gets the recommended time interval after which the client can retry
* failed requests
*
* @return the recommended time interval after which the client can retry
* failed requests.
*/
public long getRetryAfterInMilliseconds() {
long retryIntervalInMilliseconds = 0;
if (this.responseHeaders != null) {
String header = this.responseHeaders.get(HttpConstants.HttpHeaders.RETRY_AFTER_IN_MILLISECONDS);
if (StringUtils.isNotEmpty(header)) {
try {
retryIntervalInMilliseconds = Long.parseLong(header);
} catch (NumberFormatException e) {
// If the value cannot be parsed as long, return 0.
}
}
}
// In the absence of explicit guidance from the backend, don't introduce any unilateral retry delays here.
return retryIntervalInMilliseconds;
}
/**
* Gets the http status code.
*
@ -188,74 +308,6 @@ public class DocumentClientException extends Exception {
return code;
}
/**
* Gets the error code associated with the exception.
*
* @return the error.
*/
public Error getError() {
return this.error;
}
/**
* Gets the recommended time interval after which the client can retry
* failed requests
*
* @return the recommended time interval after which the client can retry
* failed requests.
*/
public long getRetryAfterInMilliseconds() {
long retryIntervalInMilliseconds = 0;
if (this.responseHeaders != null) {
String header = this.responseHeaders.get(HttpConstants.HttpHeaders.RETRY_AFTER_IN_MILLISECONDS);
if (StringUtils.isNotEmpty(header)) {
try {
retryIntervalInMilliseconds = Long.parseLong(header);
} catch (NumberFormatException e) {
// If the value cannot be parsed as long, return 0.
}
}
}
//
// In the absence of explicit guidance from the backend, don't introduce
// any unilateral retry delays here.
return retryIntervalInMilliseconds;
}
/**
* Gets the response headers as key-value pairs
*
* @return the response headers
*/
public Map<String, String> getResponseHeaders() {
return this.responseHeaders;
}
/**
* Gets the resource address associated with this exception.
*
* @return the resource address associated with this exception.
*/
String getResourceAddress() {
return this.resourceAddress;
}
/**
* Gets the Client side request statistics associated with this exception.
*
* @return Client side request statistics associated with this exception.
*/
public ClientSideRequestStatistics getClientSideRequestStatistics() {
return clientSideRequestStatistics;
}
public void setClientSideRequestStatistics(ClientSideRequestStatistics clientSideRequestStatistics) {
this.clientSideRequestStatistics = clientSideRequestStatistics;
}
@Override
public String toString() {
return getClass().getSimpleName() + "{" +
@ -280,6 +332,19 @@ public class DocumentClientException extends Exception {
return innerErrorMessage;
}
/**
* Gets the resource address associated with this exception.
*
* @return the resource address associated with this exception.
*/
String getResourceAddress() {
return this.resourceAddress;
}
void setResourceAddress(String resourceAddress) {
this.resourceAddress = resourceAddress;
}
private String getCauseInfo() {
Throwable cause = getCause();
if (cause != null) {

138
commons/src/test/java/com/microsoft/azure/cosmosdb/DocumentClientExceptionTest.java
Просмотреть файл

@ -26,12 +26,24 @@ import com.microsoft.azure.cosmosdb.internal.InternalServerErrorException;
import com.microsoft.azure.cosmosdb.internal.directconnectivity.GoneException;
import com.microsoft.azure.cosmosdb.internal.directconnectivity.RequestTimeoutException;
import com.microsoft.azure.cosmosdb.rx.internal.BadRequestException;
import io.micrometer.core.instrument.DistributionSummary;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import io.reactivex.netty.protocol.http.client.HttpResponseHeaders;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import static com.google.common.base.Strings.lenientFormat;
import static com.microsoft.azure.cosmosdb.internal.HttpConstants.StatusCodes.BADREQUEST;
@ -40,9 +52,87 @@ import static com.microsoft.azure.cosmosdb.internal.HttpConstants.StatusCodes.IN
import static com.microsoft.azure.cosmosdb.internal.HttpConstants.StatusCodes.REQUEST_TIMEOUT;
import static org.assertj.core.api.Assertions.assertThat;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.FileAssert.fail;
public class DocumentClientExceptionTest {
@Test(groups = { "unit" })
public void canEnumerateWhileUpdatingHeaders(Method method) {
final DocumentClientException dce = new DocumentClientException(0, method.getName());
final ExecutorService threadPool = Executors.newFixedThreadPool(4);
final MeterRegistry meterRegistry = new SimpleMeterRegistry();
final DistributionSummary[] summary = {
DistributionSummary.builder("responseHeaders.size").register(meterRegistry),
DistributionSummary.builder("requestHeaders.size").register(meterRegistry),
};
final List<Callable<Boolean>> callables = Arrays.asList(
() -> {
for (int i = 0; i < 1000; i++) {
final Map<String, String> headers = dce.getResponseHeaders();
final String string = headers.toString();
final double size = headers.size();
summary[0].record(size);
}
return Boolean.TRUE;
},
() -> {
for (int i = 0; i < 5000; i++) {
dce.getResponseHeaders().put("foo." + i, "bar");
}
Thread.sleep(1);
for (int i = 0; i < 5000; i++) {
dce.getResponseHeaders().remove("foo." + i);
}
return Boolean.TRUE;
},
() -> {
for (int i = 0; i < 1000; i++) {
final Map<String, String> headers = dce.getRequestHeaders();
final String string = headers.toString();
final double size = headers.size();
summary[1].record(size);
}
return Boolean.TRUE;
},
() -> {
for (int i = 0; i < 1000; i++) {
dce.setRequestHeaders(dce.getResponseHeaders());
}
return Boolean.TRUE;
});
final List<Future<Boolean>> futures;
try {
futures = threadPool.invokeAll(callables);
} catch (InterruptedException error) {
fail(lenientFormat("unexpected %s", error), error);
return;
}
for (Future<Boolean> future : futures) {
try {
assertTrue(future.get());
} catch (ExecutionException | InterruptedException error) {
fail(lenientFormat("unexpected %s", error), error);
return;
}
}
System.out.println(summary[0].takeSnapshot());
System.out.println(summary[1].takeSnapshot());
}
@Test(groups = { "unit" })
public void headerNotNull1() {
DocumentClientException dce = new DocumentClientException(0);
@ -93,6 +183,54 @@ public class DocumentClientExceptionTest {
assertThat(dce.getResponseHeaders()).contains(respHeaders.entrySet().iterator().next());
}
@Test(groups = { "unit" })
public void nullClearsRequestHeaders(Method method) {
final DocumentClientException dce = new DocumentClientException(0, method.getName());
final Map<String, String> values = new HashMap<>();
values.put("foo", "bar");
values.put("bar", "baz");
assertNotNull(dce.getRequestHeaders());
try {
dce.setRequestHeaders(values);
} catch (Throwable error) {
fail(lenientFormat("unexpected %s", error), error);
return;
}
assertThat(dce.getRequestHeaders().size()).isEqualTo(values.size());
try {
dce.setRequestHeaders(null);
} catch (Throwable error) {
fail(lenientFormat("unexpected %s", error), error);
return;
}
assertThat(dce.getRequestHeaders().size()).isZero();
}
@Test(groups = { "unit" })
public void nullValuesInRequestHeadersAreIgnored(Method method) {
final DocumentClientException dce = new DocumentClientException(0, method.getName());
final Map<String, String> values = new HashMap<>();
values.put("foo", null);
assertNotNull(dce.getRequestHeaders());
try {
dce.setRequestHeaders(values);
} catch (Throwable error) {
fail(lenientFormat("unexpected %s", error), error);
}
assertThat(dce.getRequestHeaders()).isEmpty();
}
@Test(groups = { "unit" }, dataProvider = "subTypes")
public void statusCodeIsCorrect(Class<DocumentClientException> type, int expectedStatusCode) {
try {

1
direct-impl/src/main/java/com/microsoft/azure/cosmosdb/internal/directconnectivity/rntbd/RntbdMetrics.java
Просмотреть файл

@ -26,7 +26,6 @@ package com.microsoft.azure.cosmosdb.internal.directconnectivity.rntbd;
import com.codahale.metrics.ConsoleReporter;
import com.codahale.metrics.MetricRegistry;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonPropertyOrder;
import com.google.common.net.PercentEscaper;

4
direct-impl/src/main/java/com/microsoft/azure/cosmosdb/internal/directconnectivity/rntbd/RntbdRequestRecord.java
Просмотреть файл

@ -55,10 +55,10 @@ public final class RntbdRequestRecord extends CompletableFuture<StoreResponse> {
private static final Logger logger = LoggerFactory.getLogger(RntbdRequestRecord.class);
private static final AtomicIntegerFieldUpdater REQUEST_LENGTH =
private static final AtomicIntegerFieldUpdater<RntbdRequestRecord> REQUEST_LENGTH =
AtomicIntegerFieldUpdater.newUpdater(RntbdRequestRecord.class, "requestLength");
private static final AtomicIntegerFieldUpdater RESPONSE_LENGTH =
private static final AtomicIntegerFieldUpdater<RntbdRequestRecord> RESPONSE_LENGTH =
AtomicIntegerFieldUpdater.newUpdater(RntbdRequestRecord.class, "responseLength");
private static final AtomicReferenceFieldUpdater<RntbdRequestRecord, Stage> STAGE =

2
sdk/src/test/java/com/microsoft/azure/cosmosdb/rx/internal/ConsistencyTests2.java
Просмотреть файл

@ -227,7 +227,7 @@ public class ConsistencyTests2 extends ConsistencyTestsBase {
// https://msdata.visualstudio.com/CosmosDB/_workitems/edit/355053
}
@Test(groups = {"direct"}, timeOut = 2 * CONSISTENCY_TEST_TIMEOUT)
@Test(groups = {"direct"}, timeOut = 4 * CONSISTENCY_TEST_TIMEOUT)
public void validateSessionTokenAsync() {
// Validate that document query never fails
// with NotFoundException